map与mapPartitions
map详解:
Return a new RDD by applying a function to all elements of this RDD
对RDD中的每一个元素都执行一个function
mapPartitions:
Return a new RDD by applying a function to each partition of this RDD
RDD由n个Partition构成,每个Partition由m条数据构成
map是将函数作用到每一个元素上
而mapPartitions是将函数作用到每一个分区上
如果RDD 10Partitions,每个Par存储100w
map:1000w次 connection
mapPartitions 10次 connection
但是Map也是有优势的:
普通的map操作,这一堆操作是处理那一条记录的数据(那一个元素的)
处理完之后,内存里就清掉了,空间就腾出来了。所以map一般不会涉及到OOM的
但是MapPartitions不一样,每次处理一个分区的数据,这个分区的数据处理完后,原 RDD 中分区的数据才能释放。这种情况可能导致 OOM 。如果一个分区数据太多,内存扛不住了,直接就死掉了
第一选择是MapPartitions,如果出现OOM,再考虑Map
foreach与foreachPartition同理:
在写数据到外部数据库时,优先使用foreachPartition
coalesce与repartition
coalesce 重新分区,可以选择是否进行 shuffle 过程。默认将RDD的分区数减少到指定的分区数,不能放大,多的分区数变成少的分区数,不需要数据的shuffle;如果要放大 需要将第二个参数变成true。
由于分区数决定了之后产生的文件的个数,用于合并小文件(不shuffle,从多变少)
常用于和filter算子做配合,filter过滤之后使用coalesce来减少分区。
repartition 底层是调用coalesce(num,true),能够增加或者减少分区,是肯定要进行shuffle的。
repartition用来打散数据,提高并行度,处理数据倾斜
reduceByKey与groupByKey
reduceByKey:
sc.textFile("").flatMap(_.split("\t")).map((_,1)).reduceByKey(_+_).collect
groupByKey:
sc.textFile("").flatMap(_.split("\t")).map((_,1)).groupByKey().map(x=>(x._1,x._2.sum)).collect
reduceByKey不仅简单,shuffle的数据还比groupByKey的shuffle的数据少:
原因是reduceByKey 事先在map端本地做了一次聚合操作(combiner),combiner的结果再做了shuffle,所以shuffle的数据量少一些
所以工作当中优先使用reduceByKey