http://spark.apache.org/docs/latest/rdd-programming-guide.html#shuffle-operations
Spark中的某些操作会触发称为shuffle的事件。 shuffle是Spark的重新分配数据的机制,因此它可以跨分区进行不同的分组。 这通常涉及跨执行程序和机器复制数据,使得shuffle成为复杂且昂贵的操作。
为了理解在shuffle期间发生的事情,我们可以考虑reduceByKey操作的示例。 reduceByKey操作生成一个新的RDD(重新分区),其中单个键的所有值都组合成一个元组,是 key 和 对与该key关联的所有value执行reduce函数的结果。 挑战在于,并非单个key的所有值都必须位于同一个分区,甚至是同一个机器上,但它们必须位于同一位置才能计算结果。
背景:
在Spark中,数据通常不跨分区分布,以便在特定操作的必要位置。 在计算过程中,单个任务将在单个分区上运行 - 因此,要组织单个reduceByKey reduce任务执行的所有数据,Spark需要执行全部操作。 它必须从所有分区读取以查找所有键的所有值,然后将各个值组合在一起以计算每个键的最终结果 - 这称为shuffle。
尽管新shuffle数据的每个分区中的元素集将是确定性的,并且分区本身的排序也是如此,但这些元素的排序不是。 如果在shuffle后需要可预测的有序数据,则可以使用:
mapPartitions使用例如.sorted对每个分区进行排序;
repartitionAndSortWithinPartitions在同时重新分区的同时有效地对分区进行排序;
sortBy来创建一个全局排序的RDD
可以导致shuffle的操作包括重新分区操作,例如重新分区和合并,“ByKey操作(计数除外)”,如groupByKey和reduceByKey,以及联合操作,如 cogroup and join.
性能影响:
Shuffle是一项昂贵的操作,因为它涉及磁盘I / O,数据序列化和网络I / O. 为了组织shuffle的数据,
Spark生成多组任务 -map tasks以组织数据,以及一组 reduce tasks来聚合它。 这个术语来自MapReduce,并不直接与Spark的map和reduce操作有关。
在内部,各个map任务的结果会保留在内存中,直到它们无法fit。 然后,这些基于目标分区进行排序并写入单个文件。 在reduce方面,tasks 读取相关的排序的块。
某些shuffle操作会消耗大量的堆内存,因为它们使用内存中的数据结构来在传输记录之前或之后组织记录。 具体来说,reduceByKey和aggregateByKey在map侧创建这些结构,并且'ByKey操作在reduce侧生成这些结构。 当数据不适合内存时,Spark会将这些表溢出到磁盘,从而导致磁盘I / O的额外开销和垃圾收集增加。
Shuffle还会在磁盘上生成大量中间文件。 从Spark 1.3开始,这些文件将被保留,直到不再使用相应的RDD并进行垃圾回收。 这样做是为了在重新计算 lineage 时不需要重新创建shuffle文件。 如果应用程序保留对这些RDD的引用或者GC不经常启动,则垃圾收集可能仅在很长一段时间后才会发生。 这意味着长时间运行的Spark作业可能会占用大量磁盘空间。
配置Spark上下文时,spark.local.dir配置参数指定临时存储目录。
可以通过调整各种配置参数来调整shuffle行为。See the ‘Shuffle Behavior’ section within the Spark Configuration Guide.
网友评论