Spark Shuffle原理详解

Posted 2021-09-16 <一蓑烟雨任平生>

tags:

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

（1）Shuffle概述

Shuffle 就是对数据进行重组，是把一组无规则的数据尽量转换成一组具有一定规则的数据。由于分布式计算的特性和要求，在实现细节上更加繁琐和复杂。

在 MapReduce 框架，Shuffle 是连接 Map 和 Reduce 之间的桥梁，Map 阶段通过 shuffle 读取数据并输出到对应的 Reduce。而 Reduce 阶段负责从 Map 端拉取数据并进行计算。在整个 shuffle 过程中，往往伴随着大量的磁盘和网络 I/O。所以 shuffle 性能的高低也直接决定了整个程序的性能高低。下图为 Hadoop Shuffle 过程。

Spark 也有自己的 shuffle 实现过程。在 DAG 调度的过程中，Stage 阶段的划分是根据是否有 shuffle 过程，也就是存在 ShuffleDependency 宽依赖的时候，需要进行 shuffle。 并且在划分 Stage 并构建 ShuffleDependency 的时候进行 shuffle 注册，获取后续数据读取所需要的 ShuffleHandle, 最终每一个 job 提交后都会生成一个 ResultStage 和若干个 ShuffleMapStage，其中 ResultStage 表示生成作业的最终结果所在的 Stage。ResultStage 与 ShuffleMapStage 中的 task 分别对应着 ResultTask 与 ShuffleMapTask。一个作业，除了最终的 ResultStage 外，其他若干 ShuffleMapStage 中各个 ShuffleMapTask 都需要将最终的数据根据相应的 Partitioner 对数据进行分组，然后持久化分区的数据。

（2）Hash Shuffle机制

（2.1）Hash Shuffle概述

在 spark-1.6 版本之前，一直使用 HashShuffle，在 spark-1.6 版本之后使用 Sort Shuffle，因为 HashShuffle 存在的不足所以就替换了 HashShuffle。

我们知道，Spark 的运行主要分为 2 部分：一部分是驱动程序，其核心是 SparkContext。另一部分是 Worker 节点上 Task,它是运行实际任务的。程序运行的时候，Driver 和 Executor 进程相互交互，Driver 会分配 Task 到 Executor，也就是 Driver 跟 Executor 会进行网络传输。另外，当前 Task 要抓取其他上游的 Task 的数据结果，所以这个过程中就不断的产生网络结果。其中下一个 Stage 向上一个 Stage 要数据这个过程，我们就称之为 Shuffle。

（2.2）没有优化之前的Hash Shuffle机制

HashShuffle没有优化之前的细节过程：

在 HashShuffle 没有优化之前，每一个 ShufflleMapTask 会为每一个 ReduceTask 创建一个bucket 缓存，并且会为每一个 bucket 创建一个文件。这个 bucket 存放的数据就是经过 Partitioner 操作(默认是 HashPartitioner)之后找到对应的 bucket 然后放进去，最后将数据刷新 bucket 缓存的数据到磁盘上，即对应的 block file。

然后 ShuffleMapTask 将输出作为 MapStatus 发送到 DAGScheduler 的 MapOutputTrackerMaster，每一个 MapStatus 包含了每一个 ResultTask 要拉取的数据的位置和大小。

接下来 ResultTask 去利用 BlockStoreShuffleFetcher 向 MapOutputTrackerMaster 获取 MapStatus，看哪一份数据是属于自己的，然后底层通过 BlockManager 将数据拉取过来。

拉取过来的数据会组成一个内部的 ShuffleRDD，优先放入内存，内存不够用则放入磁盘，然后 ResulTask 开始进行聚合，最后生成我们希望获取的那个 MapPartitionRDD。

自己的话总结：

每一个上游ShufflleMapTask 根据下游 ReduceTask数量，产生对应多个的bucket内存，这个bucket存放的数据是经过Partition操作（默认是Hashpartition）之后找到对应的 bucket 然后放进去，bucket内存大小默认是32k，最后将bucket缓存的数据溢写到磁盘，即为对应的block file。接下来Reduce Task底层通过 BlockManager 将数据拉取过来。拉取过来的数据会组成一个内部的 ShuffleRDD，优先放入内存，内存不够用则放入磁盘。

没有优化的Hash Shuffle的缺点

如上图所示：在这里有 1 个 worker， 2 个 executor，每一个 executor 运行 2 个 ShuffleMapTask，有三个 ReduceTask，计算方式为：executor 数量每个 executor 的 ShuffleMapTask 数量ReduceTask 数量。所以总共就有 2 * 2 * 3=12 个 bucket 以及对应 12 个 block file(分区文件)。

如果数据量较大，将会生成 M*R 个小文件，比如 ShuffleMapTask 有 100 个，ResultTask 有 100 个，这就会产生 100 * 100=10000 个小文件
bucket 缓存很重要，需要将 ShuffleMapTask 所有数据都写入 bucket，然后再刷到磁盘。那么如果 Map 端数据过多，这就很容易造成内存溢出。尽管后面有优化，bucket 写入的数据达到刷新到磁盘的阀值之后，就会将数据一点一点的刷新到磁盘，但是这样磁盘 I/O 就多了。

（2.3）优化后的Hash Shuffle机制

HashShuffle优化之后的细节过程：
每一个 Executor 进程根据核数，决定 Task 的并发数量，比如 executor 核数是 2，那就可以并发运行两个 task，如果是一个则只能运行一个 task。

假设 executor 核数是 1，ShuffleMapTask 数量是 M,那么它依然会根据 ResultTask 的数量 R，创建 R 个 bucket 缓存，然后对 key 进行 hash，数据进入不同的 bucket 中，每一个 bucket 对应着一个 block file,用于刷新 bucket 缓存里的数据。

然后下一个 task 运行的时候，就不会再创建新的 bucket 和 block file，而是复用之前的 task 已经创建好的 bucket 和 block file。即所谓同一个 Executor 进程里所有 Task 都会把相同的 key 放入相同的 bucket 缓冲区中。

这样的话，生成文件的数量就是(本地 worker 的所有 executor 对应的 cores 的总数 *ResultTask 数量)如上图所示，即 2 * 3 = 6 个文件，每一个 Executor 的 shuffleMapTask 数量 100,ReduceTask 数量即为 100。

接下来举例比较一下，未优化的 HashShuffle 的文件数是 2 * 100 * 100 =20000，优化之后的数量是 2*100 = 200 文件，相当于少了 100 倍。

自己的话总结：

每一个 Executor 进程根据核数，决定 Task 的并发数量，如果executor 核数是1，则只能运行一个task。ShuffleMapTask 会根据 ResultTask 的数量，产生对应多个的bucket内存，然后对 key 进行 hash分区，数据进入不同的 bucket 中，每一个 bucket 对应着一个 block file,用于刷新 bucket缓存里的数据。然后下一个 task 运行的时候，就不会再创建新的 bucket 和 block file，而是复用之前的 task 已经创建好的 bucket 和 block file。即所谓同一个 Executor 进程里所有 Task 都会把相同的 key放入相同的 bucket 缓冲区中。

优化过的Hash Shuffle的缺点
如果 Reducer 端的并行任务或者是数据分片过多的话则 Core * Reducer Task 依旧过大，也会产生很多小文件。

（3）Sort Shuffle机制

为了缓解 Shuffle 过程产生文件数过多和 Writer 缓存开销过大的问题，spark 引入了类似于 hadoop Map-Reduce 的 shuffle 机制。该机制每一个 ShuffleMapTask 不会为后续的任务创建单独的文件，而是会将所有的 Task 结果写入同一个文件，并且对应生成一个索引文件。 以前的数据是放在内存缓存中，等到缓存读取完数据后再刷到磁盘，现在为了减少内存的使用，在内存不够用的时候，可以将输出溢写到磁盘。结束的时候，再将这些不同的文件联合内存（缓存）的数据一起进行归并，从而减少内存的使用量。一方面文件数量显著减少，另一方面减少 Writer 缓存所占用的内存大小，而且同时避免 GC 的风险和频率。

Sort Shuffle原理：普通的SortShuffle

在普通模式下，数据会先写入一个内存数据结构中，此时根据不同的shuffle算子，可以选用不同的数据结构。如果是由聚合操作的shuffle算子，就是用map的数据结构（边聚合边写入内存），如果是join的算子，就使用array的数据结构（直接写入内存）。接着，每写一条数据进入内存数据结构之后，就会判断是否达到了某个临界值，如果达到了临界值的话，就会尝试的将内存数据结构中的数据溢写到磁盘，然后清空内存数据结构。

在溢写到磁盘文件之前，会先根据key对内存数据结构中已有的数据进行排序，排序之后，会分批将数据写入磁盘文件。默认的batch数量是10000条，也就是说，排序好的数据，会以每批次1万条数据的形式分批写入磁盘文件，写入磁盘文件是通过Java的BufferedOutputStream实现的。BufferedOutputStream是Java的缓冲输出流，首先会将数据缓冲在内存中，当内存缓冲满溢之后再一次写入磁盘文件中，这样可以减少磁盘IO次数，提升性能。

此时task将所有数据写入内存数据结构的过程中，会发生多次磁盘溢写，会产生多个临时文件，最后会将之前所有的临时文件都进行合并，最后会合并成为一个大文件。最终只剩下两个文件，一个是合并之后的数据文件，一个是索引文件（标识了下游各个task的数据在文件中的start offset与end offset）。最终再由下游的task根据索引文件读取相应的数据文件。

Sort Shuffle原理： bypass运行机制

此时上游stage的task会为每个下游stage的task都创建一个临时磁盘文件，并将数据按key进行hash然后根据key的hash值，将key写入对应的磁盘文件之中。当然，写入磁盘文件时也是先写入内存缓冲，缓冲写满之后再溢写到磁盘文件的。最后，同样会将所有临时磁盘文件都合并成一个磁盘文件，并创建一个单独的索引文件。

bypass机制与普通SortShuffleManager运行机制的不同在于：

a、磁盘写机制不同；
b、不会进行排序。

也就是说，启用该机制的最大好处在于，shuffle write过程中，不需要进行数据的排序操作，也就节省掉了这部分的性能开销。

自己的话总结：
普通的SortShuffle机制

在普通模式下，数据会先写入一个内存数据结构中，如果是由聚合操作的shuffle算子用map数据结构，如果是join算子就用Array数据结构。在写入的过程中如果达到了临界值，就会将内存数据结构中的数据溢写到磁盘，然后清空内存数据结构。在溢写磁盘之前，会先根据key对内存数据结构中的数据进行排序，排序好的数据，会以每批次1万条数据的形式分批写入磁盘文件。在task将所有数据写入内存数据结构的过程中，会发生多次磁盘溢写，会产生多个临时文件，最后会将之前所有的临时文件都进行合并，最后会合并成为一个大文件。最终只剩下两个文件，一个是合并之后的数据文件，一个是索引文件，最终再由下游的task根据索引文件读取相应的数据文件。

bypass运行机制

bypass的就是不排序，还是用hash去为key分磁盘文件，分完之后再合并，形成一个索引文件和一个合并后的key hash文件。省掉了排序的性能。

Sort Shuffle 有几种不同的策略：

BypassMergeSortShuffleWriter（Bypass 机制）
SortShuffleWriter（普通机制）
UnsafeShuffleWriter

对于 BypassMergeSortShuffleWriter，使用这个模式的特点为：

主要用于处理不需要排序和聚合的 Shuffle 操作，所以数据是直接写入文件，数据量较大的时候，网络 I/O 和内存负担较重。
主要适合处理 Reducer 任务数量比较少的情况。
将每一个分区写入一个单独的文件，最后将这些文件合并,减少文件数量。但是这种方式需要并发打开多个文件，对内存消耗比较大。

因为 BypassMergeSortShuffleWriter 这种方式比 SortShuffleWriter 更快，所以如果在 Reducer 数量不大，又不需要在 map 端聚合和排序，而且 Reducer 的数目小于 spark.shuffle.sort.bypassMergeThreshold 指定的阀值（默认 200）时，就是用的是这种方式（即启用条件）。

对于 SortShuffleWriter(普通机制),使用这个模式的特点为：

比较适合数据量很大的场景或者集群规模很大。
引入了外部排序器，可以支持在 Map 端进行本地聚合或者不聚合。
如果外部排序器 enable 了 spill 功能，如果内存不够，可以先将输出溢写到本地磁盘，最后将内存结果和本地磁盘的溢写文件进行合并。

另外，这个 Sort-Based Shuffle 跟 Executor 核数没有关系，即跟并发度没有关系，它是每一个 ShuffleMapTask 都会产生一个 data 文件和 index 文件，所谓合并也只是将该 ShuffleMapTask 的各个 partition 对应的分区文件合并到 data 文件而已。所以这个就需要和 Hash-BasedShuffle 的 consolidation 机制区别开来。

（4）Spark Shuffle调优

调节map端缓冲区大小
- map端缓冲的默认配置是32KB
- val conf = new SparkConf().set("spark.shuffle.file.buffer", "64")
调节reduce端拉取数据缓冲区大小
- 如果内存资源较为充足，适当增加拉取数据缓冲区的大小，可以减少拉取数据的次数，也就可以减少网络传输的次数，进而提升性能
- reduce端数据拉取缓冲区大小默认48MB
- val conf = new SparkConf().set("spark.reducer.maxSizeInFlight", "96")
调节reduce端拉取数据重试次数
- reduce端拉取数据重试次数默认为3
- val conf = new SparkConf().set("spark.shuffle.io.maxRetries", "6")
调节reduce端拉取数据等待间隔
- reduce端拉取数据等待间隔默认为5s
- val conf = new SparkConf().set("spark.shuffle.io.retryWait", "60s")
调节SortShuffle排序操作阈值
- SortShuffleManager排序操作阈值默认为200
- val conf = new SparkConf().set("spark.shuffle.sort.bypassMergeThreshold", "400")