学习笔记Spark—— Spark编程基础(创建RDDRDD算子文件读取与存储)

Posted 别呀

tags:

篇首语:本文由小常识网(cha138.com)小编为大家整理,主要介绍了学习笔记Spark—— Spark编程基础(创建RDDRDD算子文件读取与存储)相关的知识,希望对你有一定的参考价值。

一、创建RDD

1.1、启动Spark shell

进入Spark命令行交互界面spark-shell
退出交互界面:q


查看客户端:(http://master:8080


设置日志级别

sc.setLogLevel("INFO")
sc.setLogLevel("WARN")

1.2、创建RDD

在Spark中创建RDD的创建方式大概可以分为三种:

  • 从集合中创建RDD
  • 从外部存储创建RDD
  • 从其他RDD创建

1.2.1、从集合中创建RDD

parallelize():通过parallelize函数把一般数据结构加载为RDD
parallelize[T: ClassTag](seq: Seq[T],numSlices: Int = defaultParallelism): RDD[T]

Parallelize Rdd默认分区数:sc.defaultParallelism,可通过spark.default.parallelism设置sc.defaultParallelism的值,没有配置spark.default.parallelism时的默认值等于cpu的核数

例1

例2


1.2.2、从外部存储中创建RDD

通过textFile直接加载数据文件为RDD

textFile(path: String, minPartitions: Int = defaultMinPartitions): RDD[String]

读取HDFS时默认分区数:rdd的分区数 = max(hdfs文件的block数目, sc.defaultMinPartitions),sc.defaultMinPartitions=min(sc.defaultParallelism,2)

从本地文件读取:本地file的分片规则,应该按照hdfs的block大小划分,但实测的结果是固定按照32M来分片

读取HDFS上文件:

读取本地文件:
(注意:读取本地文件时,要确保每个集群上都要有文件,否则会报错)


任务1:

1、HDFS上有三份文件,分别为student.txt(学生信息表),result_bigdata.txt(大数据基础成绩表),result_math.txt(数学成绩表)
加载student.txt为名称为student的RDD数据,result_bigdata.txt为名称为bigdata的RDD数据,result_math.txt为名称为math的RDD数据

数据:

程序:


二、RDD算子

2.1、map与flatMap算子应用

2.1.1、map

map(func)
  • Transformation类型算子
  • map: 将原来RDD的每个数据项通过map中的用户自定义函数f转换成一个新的RDD,map操作不会改变RDD的分区数目

示例:使用map函数对RDD中每个元素进行倍数操作

2.1.2、flatMap

flatMap(func)
  • Transformation类型算子
  • flatMap:对集合中的每个元素进行map操作再扁平化

示例:使用flatMap分割单词


2.1.3、mapPartitions

mapPartitions(func)
  • Transformation类型算子
  • 和map功能类似,但是输入的元素是整个分区,即传入函数的操作对象是每个分区的Iterator集合,该操作不会导致Partitions数量的变化

示例:取出每个分区中大于3的值


2.2、sortBy与filter算子应用

2.2.1、sortBy

sortBy(f:(T) => K, ascending, numPartitions)
  • Transformation类型算子
  • 是可以对标准RDD进行排序
  • sortBy()可接受三个参数:
  • f:(T) => K:左边是要被排序对象中的每一个元素,右边返回的值是元素中要进行排序的值。
  • ascending:决定排序后RDD中的元素是升序还是降序,默认是true,也就是升序,false为降序排序。
  • numPartitions:该参数决定排序后的RDD的分区个数,默认排序后的分区个数和排序之前的个数相等。

示例:按照每个元素的第二个值进行降序排序


2.2.2、filter

filter(func)
  • Transformation类型算子
  • 保留通过函数func,返回值为true的元素,组成新的RDD
  • 过滤掉data RDD中元素小于或等于2的元素

示例:


任务2:

根据任务1得到的RDD bigdata及math,取出成绩排名前5的学生成绩信息



2.3、交集与并集计算的算子应用

2.3.1、distinct

distinct([numPartitions]))
  • Transformation类型算子
  • 针对RDD中重复的元素,只保留一个元素

示例:


2.3.2、union

union(otherDataset)
  • 合并RDD,需要保证两个RDD元素类型一致

示例:合并rdd1和rdd2


2.3.3、intersection

intersection(otherDataset)
  • 找出两个RDD的共同元素,也就是找出两个RDD的交集

示例:找出c_rdd1和c_rdd2中相同的元素


2.3.4、subtract

subtract (otherDataset)
  • 获取两个RDD之间的差集

示例:找出rdd1与rdd2之间的差集


2.3.5、cartesian

cartesian(otherDataset)
  • 笛卡尔积就是将两个集合的元素两两组合成一组

示例:


任务3:

1、找出考试成绩得过100分的学生ID,最终的结果需要集合到一个RDD中。
2、找出两门成绩都得100分的学生ID,结果汇总为一个RDD。


2.4、 键值对RDD常用算子

虽然大部分Spark的RDD操作都支持所有种类的单值RDD,但是有少部分特殊的操作只能作用于键值对类型的RDD。

顾名思义,键值对RDD由一组组的键值对组成,这些RDD被称为PairRDD。PairRDD提供了并行操作各个键或跨节点重新进行数据分组的操作接口。例如,PairRDD提供了reduceByKey()方法,可以分别规约每个键对应的数据,还有join()方法,可以把两个RDD中键相同的元素组合在一起,合并为一个RDD。


2.4.1、 创建键值对RDD

将一个普通的RDD转化为一个PairRDD时可以使用map函数来进行操作,传递的函数需要返回键值对。

做为键值对类型的RDD,包含了键跟值两个部分。Spark提供了两个方法分别获取键值对RDD的键跟值。keys返回一个仅包含键的RDD,values返回一个仅包含值的RDD。


2.4.2、mapValues

mapValues(func)
  • 类似map,针对键值对(Key,Value)类型的数据中的Value进行map操作,而不对Key进行处理

示例:


2.4.3、groupByKey

groupByKey([numPartitions])
  • 按键分组,在(K,V)对组成的RDD上调用时,返回(K,Iterable)对组成的新的RDD。

示例:将rdd按键进行分组


2.4.4、reduceByKey

  • 将键值对RDD按键分组后进行聚合
  • 当在(K,V)类型的键值对组成的RDD上调用时,返回一个(K,V)类型键值对组成的新RDD
  • 其中新RDD每个键的值使用给定的reduce函数func进行聚合,该函数必须是(V,V)=>V类型

示例:统计每个键出现的次数


2.4.5、join

  • 把键值对数据相同键的值整合起来
  • 其他连接有:leftOuterJoin, rightOuterJoin, and fullOuterJoin

join: 把键值对数据相同键的值整合起来


2.5、常用Action类型算子

2.5.1、lookup

lookup(key: K)
  • Action类型算子
  • 作用于(K,V)类型的RDD上,返回指定K的所有V值

示例:


2.5.2、collect

collect()
  • 返回RDD中所有的元素
  • collectAsMap(): Map[K, V]

示例:


2.5.3、take

take(num)
  • 返回RDD前面num条记录

示例:


2.5.4、count

count()
  • 计算RDD中所有元素个数


任务4:

1、输出每位学生的总成绩,要求将两个成绩表中学生ID相同的成绩相加。
2、输出每位学生的平均成绩,要求将两个成绩表中学生ID相同的成绩相加并计算出平均分。
3、合并每个学生的总成绩和平均成绩。





三、文件读取与存储

3.1、saveAsTextFile

saveAsTextFile(path: String)
  • 把RDD保存到HDFS中



3.2、repartition

repartition(numPartitions: Int)
  • 可以增加或减少此RDD中的并行级别。在内部,它使用shuffle重新分发数据。
  • 如果要减少此RDD中的分区数,请考虑使用coalesce,这样可以避免执行shuffle。
  • coalesce(numPartitions: Int, shuffle: Boolean = false, partitionCoalescer: Option[PartitionCoalescer] = Option.empty)



3.3、saveAsSequenceFile

saveAsSequenceFile(path)
  • 保存成序列化文件
  • 将数据集的元素作为Hadoop SequenceFile编写,只支持键值对RDD


3.4、sequenceFile

sequenceFile[K, V](path: String, keyClass: Class[K], valueClass: Class[V], minPartitions: Int)
  • 读取序列化文件


综合练习1

综合练习:基于3个基站的日志数据,要求计算某个手机号码在一天之内出现时间最多的两个地点。

模拟了一些简单的日志数据,共4个字段:手机号码,时间戳,基站id,连接类型(1表示建立连接,0表示断开连接):
基站A:

基站B:
基站C:

程序:

以上是关于学习笔记Spark—— Spark编程基础(创建RDDRDD算子文件读取与存储)的主要内容,如果未能解决你的问题,请参考以下文章

2022年Spark基础学习笔记

Spark基础学习笔记28:Spark SQL数据源 - JDBC

Spark基础学习笔记30:Spark SQL案例分析

Spark基础学习笔记29:Spark SQL内置函数

Spark基础学习笔记16:创建RDD

Spark基础学习笔记26:Spark SQL数据源 - JSON数据集