Spark Shuffle 机制解析

Posted 2021-10-08 东海陈光剑

tags:

篇首语：本文由小常识网(cha138.com)小编为大家整理，主要介绍了Spark Shuffle 机制解析相关的知识，希望对你有一定的参考价值。

1. 什么是 Shuffle

当一个父 RDD 分区的数据分散到了多个子 RDD 的分区中时，这时会产生 Shuffle，即宽依赖之间会有 Shuffle。

Reduce Task 去拉取 Map Task 数据的时候会产生大量的网络、磁盘 IO、内存的消耗，Shuffle 性能的高低对整体任务的性能影响很大。
Shuffle 通常分为两个阶段，Map 阶段数据的准备及划分，Reduce 阶段数据的拉取。Map 端的 Shuffle 通常被称作 Shuffle Write，Reduce 端的 Shuffle 通常被称作 Shuffle Read。

Spark 1.2 之前 Shuffle 使用的计算引擎是 HashShuffleManager，这种方式虽然快速，但是会产生大量的文件，如果有 M 个 Mapper，N 个 Reducer 就会产生 M * N 个文件，如果 Mapper 和 Reducer 的数量很庞大将会带来性能上的影响。
Spark 1.2 之后引入了 SortShuffleManager，这种方式 Reduce 端需要读取的文件更少，因为 Map 端的每个 Task 最后会将临时文件合并成一个文件，并且会对 Map 端的数据进行排序，并生成记录数据位置的索引文件，Reducer 可以通过索引找到自己要拉取的数据，它也是 Spark 默认使用的 Shuffle 管理器。
Spark 2.0 之后移除了 HashShuffleManager，目前 ShuffleManager 只有 SortShuffleManager 这一个实现类。

SortShuffleManager 有两种运行机制，一种是普通机制，另一种是 bypass 机制。

Shuffle Write 阶段会先将数据写入内存数据结构中，如果是聚合类型的算子 (reduceByKey)，采用 Map 数据结构，先用 Map 进行预聚合处理，再写入内存中；如果是普通的 shuffle 算子的话 (join)，采用 Array 数据结构，直接写入内存。
当内存达到阈值的时候，会将这些内存中的数据进行排序，然后分批次写入磁盘文件 (默认 1W/批)，这里并不会直接写入磁盘，会先写入内存缓冲流中，当缓冲流满溢之后，写入磁盘文件。
最后会将每个 Task 写出的文件进行合并，最终生成一份数据文件和一份索引文件，索引文件记录了 Shuffle Read 阶段每个 Task 要读取的数据在文件中对应的开始和结束位置。

普通机制

Shuffle Write 阶段会对每个 Task 数据的 key 进行 hash，相同 hash 的 key 会被写入同一个内存缓冲区，缓冲区满溢后会写到磁盘文件中。
最后会将每个 Task 写出的磁盘文件进行合并，并创建索引文件 (与普通机制下的索引文件作用相同)。

bypass 机制

shuffle read task 的数量小于等于 spark.shuffle.sort.bypassMergeThreshold 参数设置的阈值的时候，默认是 200。
触发 shuffle 的算子不能是聚合类算子，比如 reduceByKey(在 Map 端对每一个 Task 的 key 会先进行一次预聚合处理)。

在将数据写入内存缓冲区的时候，普通机制要先将数据写入 Map 或者 Array 的内存数据结构中，而 bypass 机制是根据 key 的 hash 值直接写入内存缓冲区中。
bypass 机制在写入内存缓冲区之前没有对数据的排序操作，因此在 reduce task 比较少的情况下，开启 bypass 机制，不需要对数据排序，节省运算性能。

以上是关于Spark Shuffle 机制解析的主要内容，如果未能解决你的问题，请参考以下文章