Spark

it2022-05-05  126

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

两个算子底层都是调用的combineByKeyWithClassTag。 reduceByKey不仅简单,shuffle的数据还比groupByKey的shuffle的数据少: 原因是reduceByKey 事先在map端本地做了一次聚合操作(combiner),combiner的结果再做了shuffle,所以shuffle的数据量少一些 groupByKey可以实现分区间的函数的自定义,分区内函数也可以自定义,功能上灵活一些,不像reduceByKey直接把它写死了

排序算子的区别

Order By 是全局排序的,是只有一个reduce处理的; Sort By 是分区排序的,只能实现分区内有序,不能实现全局排序; Distributed By不是排序,是按照后面的字段来分发数据; Cluster By是Sort By和Distributed By合起来的;


最新回复(0)