。

这里我们分析其底层逻辑#xff0c;结合源码和原理来解释为什么会有

Shuffle

是一个排序算子虽然它在某些场景下可能看起来是分区内排序但实际上在需要全局排序时会触发

Shuffle。

这里我们分析其底层逻辑结合源码和原理来解释为什么会有

Shuffle

重新分区Repartition确保每个分区中的数据按照全局范围内的排序键正确分布每个分区内部再完成排序。

Shuffle因为数据需要从一个分区转移到另一个分区以保证全局有序性。

关键点

repartitionAndSortWithinPartitions

sortBy

repartitionAndSortWithinPartitions

this.keyBy(f).repartitionAndSortWithinPartitions(new

RangePartitioner(numPartitions,

this,

ascending))(ordInverse).valueskeyBy(f)

将数据转化为

repartitionAndSortWithinPartitions

Shuffle

当目标分区数量与当前分区数量不一致时用户指定分区数或默认分区数会触发

Shuffle即使目标分区数一致只要需要保证全局有序也需要重新分布数据来确保各分区内数据按键范围划分。

Shuffle

全局排序分区间重新分布数据确保所有分区的排序键范围是连续的。

负载均衡通过

RangePartitioner

分布数据避免某些分区过大或过小的问题。

分区内排序确保每个分区内部数据按键排序。

源码分析

repartitionAndSortWithinPartitions

def

repartitionAndSortWithinPartitions(partitioner:

ord:

partitioner)shuffled.setKeyOrdering(ord)new

(context,

Some(ord))sorter.insertAll(iter)context.taskMetrics().incMemoryBytesSpilled(sorter.memoryBytesSpilled)context.taskMetrics().incDiskBytesSpilled(sorter.diskBytesSpilled)context.taskMetrics().incPeakExecutionMemory(sorter.peakMemoryUsedBytes)sorter.iterator})

}ShuffledRDD

对每个分区内的数据进行排序如果数据超出内存会使用磁盘作为临时存储。

举例说明

的参数设置。

如果不显式指定目标分区数numPartitions

参数sortBy

println(sortedRdd.collect().mkString(,

))初始数据分区

partition.toList.sorted.iterator)

sorted.collect().foreach(println)初始数据分区

1[5,

Shuffle这是为了重新分区以确保分区范围和分区内排序。

如果只需要分区内排序mapPartitions

Shuffle

会显著增加网络传输和计算成本。

如无必要尽量避免全局排序优先考虑局部排序或

Top-N