算子的分类和 宽依赖算子、窄依赖算子

一、转换算子转换算子

textfile,也会惰性加载

Transformation,懒执行,需要Action触发执行
filter
过滤 RDD[T]==>RDD[T],窄依赖

map
RDD[T] ->RDD[O], 窄依赖

flatMap
RDD[T]–>RDD[[O]],一对多 ,窄依赖,

mapToPair
sample
抽样算子 RDD[T]–>RDD[O],窄依赖

sortBy
RDD[T]–>RDD[T], 根据你指定的内容排序 宽依赖

sortByKey
根据你的K排序,要求RDD中必须是KV的,宽依赖

reduceByKey
根据RDD的K分组之后聚合(累加,字符串连接) , 宽依赖

join
把两个RDD根据K相同合并,结果RDD[K,(V1,V2)] ,宽依赖

leftOuterJoin
左连接 和下面的一致 都是宽依赖

rightOuterJoin
fullOuterJoin
union
把两个RDD直接聚合成一个RDD,数据不合并 ,窄依赖

intersection
取两个RDD的交集,窄依赖

subtract
mapPartitions
把整个分区的数据作为一个迭代器一次计算 数据量不是特别大 会有性能提升,窄依赖

distinct
去重算子 本质是map + reduceByKey+map 宽依赖

cogroup
(K,V) (K,W)=>(K,([V],[W])) RDD1相同的key value 放在[V]中 另一个RDD相同的key 的value 放在[W]中 宽依赖

mapPartitionsWithIndex
把整个分区的数据作为一个迭代器一次计算 多了一个分区的index 数据量不是特别大 会有性能提升,窄依赖

repartition
可以增多分区,可以减少分区,有shuffle 宽依赖
repartition = coalesce(num,shuffle=true)

coalesce
可以增多分区,也可以减少分区,默认没有shuffle 有shuffle就 宽依赖 没shuffle 就是窄依赖
若RDD由少的分区分到多的分区时,不让产生shuffle, 不起作用
少 - > 多 false RDD1 a、b分区 RDD2 0:a 1:b 2: 窄依赖 。 true RDD1 a、b 分区 RDD2 0:a1 b1 / 1: a2 b2 / 3: a3 b3 宽依赖
多 - > 少 false RDD1 a、b、c 分区 RDD2: 0: a 、 1: b c 窄依赖 。 true RDD1 a、b、c 分区 RDD2 0: a1、b1、c1 1:a2、b2、c2 宽依赖

zip
两个RDD压成一个RDD 窄依赖

zipWithIndex
groupByKey
(K,V)–>(K,iter),根据K相同分组,分组之后把一组的V封装成一个迭代器, 宽依赖

二、行动算子Action

saveAsTextFile等存储算子,也会立即执行

触发transformation类算子执行,一个application中有一个action算子就有一个job
Action算子
foreach
循环出值

count
结果会拿到Driver端

collect
将结果拿回Driver端 返回一个列表

first
取出第一个值

take(num)
取出num个值 返回driver端

foreachPartition
reduce
聚合数据 不过 这个是在driver端 不合适聚合大量的数据 适合聚合结果数据

countByKey
countByValue

三、持久化算子

cache
默认将数据存储在内存中
cache() = persist() = persist(StorageLevel.MEMORY_ONLY)

persist
可以手动指定持久化级别
MEMORY_ONLY
MEMORY_ONLY_SER
MEMORY_AND_DISK
MEMORY_AND_DISK_SER
“_2” 是由副本
尽量少使用DISK_ONLY级别
checkpoint
将数据直接持久化到指定的目录,当lineage 计算非常复杂,可以尝试使用checkpoint ,checkpoint还可以切断RDD的依赖关系
特殊场景使用checkpoint,对RDD使用checkpoint要慎用
checkpoint要指定目录,可以将数据持久化到指定的目录中,当application执行完成之后,这个目录中的数据不会被清除

checkpoint的执行流程
1.当sparkjob执行完成之后,Spark 会从后往前回溯,找到checkpointRDD做标记
2.回溯完成之后,Spark框架会重新启动一个job,计算标记的RDD的数据,放入指定的checkpoint目录中
3.数据计算完成,放入目录之后,会切断RDD的依赖关系,当SparkApplication执行完成之后,数据目录中的数据不会被清除
优化:对哪个RDD进行checkpoint,最好先cache下,这样回溯完成后再计算这个CheckpointRDD数据的时候可以直接在内存中拿到放指定的目录中
cache和persist的注意
1.cache,persist,checkpoint 都是懒执行,最小持久化单位是partition
2.cache和persist之后可以直接赋值给一个值,下次直接使用这个值,就是使用的持久化的数据
3.如果采用第二种方式,后面不能紧跟action算子
4.cache和persist的数据,当application执行完成之后会自动清除
————————————————
版权声明:本文为CSDN博主「狂躁的辣条」的原创文章,遵循 CC 4.0 BY-SA 版权协议,转载请附上原文出处链接及本声明。
原文链接:https://blog.csdn.net/weixin_42712704/article/details/89534037

原文地址:https://www.cnblogs.com/carsonwuu/p/11596815.html