Spark Shuffle原理详解

「爱情、让人受尽委屈。」 2023-01-22 06:49 33阅读 0赞

### 目录 ###

*   *  （1）Shuffle概述
     *  （2）Hash Shuffle机制
     *   *  （2.1）Hash Shuffle概述
         *  （2.2）没有优化之前的Hash Shuffle机制
         *  （2.3）优化后的Hash Shuffle机制
     *  （3）Sort Shuffle机制
     *  （4）Spark Shuffle调优

## （1）Shuffle概述 ##

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

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

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80NTM2NjQ5OQ_size_16_color_FFFFFF_t_70]

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80NTM2NjQ5OQ_size_16_color_FFFFFF_t_70 1]

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 对数据进行分组，然后持 久化分区的数据。

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80NTM2NjQ5OQ_size_16_color_FFFFFF_t_70 2]

## （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机制 ###

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80NTM2NjQ5OQ_size_16_color_FFFFFF_t_70 3]  
**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机制 ###

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80NTM2NjQ5OQ_size_16_color_FFFFFF_t_70 4]  
**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 的风险和频率。**

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80NTM2NjQ5OQ_size_16_color_FFFFFF_t_70 5]

**Sort Shuffle原理： 普通的SortShuffle**  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80NTM2NjQ5OQ_size_16_color_FFFFFF_t_70 6]

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

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

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

**Sort Shuffle原理： bypass运行机制**  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80NTM2NjQ5OQ_size_16_color_FFFFFF_t_70 7]

此时上游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")`

--------------------

以上内容仅供参考学习，如有侵权请联系我删除！  
如果这篇文章对您有帮助，左下角的大拇指就是对博主最大的鼓励。  
您的鼓励就是博主最大的动力！

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80NTM2NjQ5OQ_size_16_color_FFFFFF_t_70]: /images/20221020/a8c87a108e4045e78be295be8f8aef68.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80NTM2NjQ5OQ_size_16_color_FFFFFF_t_70 1]: /images/20221020/a174c18b44ce411381945bd3712c90a3.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80NTM2NjQ5OQ_size_16_color_FFFFFF_t_70 2]: /images/20221020/8bc3a65900a142da9d5839ad3bbe4ec4.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80NTM2NjQ5OQ_size_16_color_FFFFFF_t_70 3]: /images/20221020/fe096ca7ef364850bbb09e7023b2d24d.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80NTM2NjQ5OQ_size_16_color_FFFFFF_t_70 4]: /images/20221020/7cd2495034954775a94869f51e8a4b1b.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80NTM2NjQ5OQ_size_16_color_FFFFFF_t_70 5]: /images/20221020/21fa71ebc2c248fbbca86a6140fcd8f4.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80NTM2NjQ5OQ_size_16_color_FFFFFF_t_70 6]: /images/20221020/d4ac719e45434e4fb871169f89551150.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80NTM2NjQ5OQ_size_16_color_FFFFFF_t_70 7]: /images/20221020/344591fdf44d4880a9980987286620a3.png