Spark学习——排序Shuffle

2,263 阅读3分钟

其他更多java基础文章:
java基础学习(目录)


这部分能力有限,所以推荐一些大神文章阅读学习:

总结

我在以我的理解简单的概括下,如有不对,希望大家及时斧正:

  • Hash Shuffle是Spark 1.2之前的默认Shuffle实现,并在Spark 2.0版本中被移除。HashShuffle有个巨大的缺点,Shuffle前在磁盘上会产生海量的小文件,此时会产生大量耗时低效的 IO 操作
  • Consolidated Hash Shuffle是Hash Shuffle的优化版,会只产生Cores数量 x Reduce端数量的小文件。
  • Sort-Based Shuffle目前默认的Shuffle实现,Sorted-Based Shuffle 会把Mapper 中每个ShuffleMapTask 所有的输出数据Data 只写到一个文件中。它会产生一个 Data 文件和一个 Index 文件,其中 Data 文件是存储当前 Task 的 Shuffle 输出的, 而 Index 文件则存储了 Data 文件中的数据通过 Partitioner 的分类信息,此时下一个阶段的 Stage 中的 Task 就是根据这个 Index 文件获取自己所需要抓取的上一个 Stage 中 ShuffleMapTask 所产生的数据;默认情况下的Sort-Based Shuffle是会在溢写磁盘前,先根据key进行排序
  • bypass模式的Sort-Based Shuffle。spark.shuffle.sort.bypassMergeThreshold 默认值为200 ,如果shuffle map task的数量小于这个阀值200,且不是聚合类的shuffle算子(比如reduceByKey),则不会进行排序。 该机制与sortshuffle的普通机制相比,在map task不多的情况下,首先写的机制是不同,其次不会进行排序。这样就可以节约一部分性能开销。
  • Tungsten-sort Based Shuffle:引入新的内存管理模型Page。个人理解是只对类似于指针或者index的二进制数据进行排序,所以会比Sort-Based Shuffle的针对java objects的排序更快。 当且仅当下面条件都满足时,才会使用新的Shuffle方式:
    • Shuffle dependency 不能带有aggregation 或者输出需要排序
    • Shuffle 的序列化器需要是 KryoSerializer 或者 Spark SQL's 自定义的一些序列化方式.
    • Shuffle 文件的数量不能大于 16777216
    • 序列化时,单条记录不能大于 128 MB