文章目录
一、创建RDD1.1、启动Spark shell1.2、创建RDD1.2.1、从集合中创建RDD1.2.2、从外部存储中创建RDD任务1:二、RDD算子2.1、map与flatMap算子应用2.1.1、map2.1.2、flatMap2.1.3、mapPartitions2.2、sortBy与filter算子应用2.2.1、sortBy2.2.2、filter任务2:2.3、交集与并集计算的算子应用2.3.1、distinct2.3.2、union2.3.3、intersection2.3.4、subtract2.3.5、cartesian任务3:2.4、 键值对RDD常用算子2.4.1、 创建键值对RDD2.4.2、mapValues2.4.3、groupByKey2.4.4、reduceByKey2.4.5、join2.5、常用Action类型算子2.5.1、lookup2.5.2、collect2.5.3、take2.5.4、count任务4:三、文件读取与存储3.1、saveAsTextFile3.2、repartition3.3、saveAsSequenceFile3.4、sequenceFile综合练习1一、创建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函数把一般数据结构加载为RDDparallelize[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 fullOuterJoinjoin: 把键值对数据相同键的值整合起来
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:
程序: