Spark：shuffle机制分析

川长思鸟来 2022-09-10 05:10 196阅读 0赞

**在Spark中，什么情况下，会发生shuffle?**  
reduceByKey、groupByKey、sortByKey、countByKey、join等操作都会产生shuffle。

那下面我们来详细分析一下Spark中的shuffle过程。  
Spark的shuffle历经了几个过程

1.  Spark 0.8及以前 使用Hash Based Shuffle
2.  Spark 0.8.1 为Hash Based Shuffle引入File Consolidation机制
3.  Spark1.6之后使用Sort-Base Shuffle，因为Hash Based Shuffle存在一些不足所以就把它替换掉了。

所以Spark Shuffle 一共经历了这几个过程:

1.  未优化的 Hash Based Shuffle
2.  优化后的Hash Based Shuffle
3.  Sort-Based Shuffle

#### 未优化的Hash Based Shuffle ####

来看一个图，假设我们是在执行一个reduceByKey之类的操作，此时就会产生shuffle  
shuffle里面会有两种task，一种是shuffleMapTask，负责拉取前一个RDD中的数据，还有一个 ResultTask，负责把拉取到的数据按照规则汇总起来  
![在这里插入图片描述][watermark_type_ZHJvaWRzYW5zZmFsbGJhY2s_shadow_50_text_Q1NETiBA566A5Y2V6ZqP6aOO_size_20_color_FFFFFF_t_70_g_se_x_16]  
1:假设有1个节点，这个节点上有2个CPU，上面运行了4个ShuffleMapTask，这样的话其实同时只有2个ShuffleMapTask是并行执行的，因为一个cpu core同时只能执行一个ShuffleMapTask。  
2:每个ShuffleMapTask都会为每个ResultTask创建一份Bucket缓存，以及对应的ShuffleBlockFile磁盘文件  
这样的话，每一个ShuffleMapTask都会产生4份Bucket缓存和对应的4个ShuffleBlockFile文件，分别对ResultTask进行处理  
3:假设另一个节点上面运行了4个ResultTask现在等着获取ShuffleMapTask的输出数据，来完成比如 ReduceByKey的操作

这是这个流程，注意了，如果有100个MapTask，100个ResultTask，那么会产生10000个本地磁盘文件，这样需要频繁的磁盘IO，是比较影响性能的。

注意，那个bucket缓存是非常重要的，ShuffleMapTask会把所有的数据都写入Bucket缓存之后，才会刷写到对应的磁盘文件中，但是这就有一个问题，如果map 端数据过多，那么很容易造成内存溢出，所以spark在优化后的Hash Based Shuffle中对这个问题进行了优化，默认这个内存缓存是100kb，当Bucket中的数据达到了阈值之后，就会将数据一点一点地刷写到对应的ShuffleBlockFile磁盘中了。  
这种操作的优点，是不容易发生内存溢出。缺点在于，如果内存缓存过小的话，那么可能发生过多的磁盘 io操作。所以，这里的内存缓存大小，是可以根据实际的业务情况进行优化的。

#### 优化后的Hash Based Shuffle ####

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

看这个优化后的shuffle流程  
1:假设机器上有2个cpu，4个shuffleMaptask，这样同时只有2个在并行执行  
2:在这个版本中，Spark引入了consolidation机制，一个ShuffleMapTask将数据写入ResultTask数量的本地文件中，这个是不变的，但是当下一个ShuffleMapTask运行的时候，可以直接将数据写入之前产生的本地文件中，相当于对多个ShuffleMapTask的输出进行了合并，从而大大减少了本地磁盘中文件的数量。

此时文件的数量变成了CPU `core数量 * ResultTask数量`，比如每个节点上有2个CPU，有100个 ResultTask，那么每个节点上会产生200个文件  
这个时候文件数量就变得少多了。

但是如果 ResultTask端的并行任务过多的话则 `CPU core * Result Task` 依旧过大，也会产生很多小文件

#### Sort-Based Shuffle ####

引入 `Consolidation` 机制虽然在一定程度上减少了磁盘文件数量，但是不足以有效提高 Shuffle 的性能， 这种情况只适合中小型数据规模的数据处理。  
为了让 Spark 能在更大规模的集群上高性能处理大规模的数据，因此 Spark 引入了 `Sort-Based Shuffle`。  
![在这里插入图片描述][watermark_type_ZHJvaWRzYW5zZmFsbGJhY2s_shadow_50_text_Q1NETiBA566A5Y2V6ZqP6aOO_size_20_color_FFFFFF_t_70_g_se_x_16 2]  
该机制针对每一个 ShuffleMapTask 都只创建一个文件，将所有的 ShuffleMapTask 的数据都写入同一个文件，并且对应生成一个索引文件。  
以前的数据是放在内存中，等到数据写完了再刷写到磁盘，现在为了减少内存的使用，在内存不够用的时候，可以将内存中的数据溢写到磁盘，结束的时候，再将这些溢写的文件联合内存中的数据一起进行归并，从而减少内存的使用量。一方面文件数量显著减少，另一方面减少缓存所占用的内存大小，而且同时避免 GC 的风险和频率。

[watermark_type_ZHJvaWRzYW5zZmFsbGJhY2s_shadow_50_text_Q1NETiBA566A5Y2V6ZqP6aOO_size_20_color_FFFFFF_t_70_g_se_x_16]: /images/20220829/593809a36d864f74bbcd47f4ab140595.png
[watermark_type_ZHJvaWRzYW5zZmFsbGJhY2s_shadow_50_text_Q1NETiBA566A5Y2V6ZqP6aOO_size_20_color_FFFFFF_t_70_g_se_x_16 1]: /images/20220829/34f8651560644ba4b26ae1f49bd50df4.png
[watermark_type_ZHJvaWRzYW5zZmFsbGJhY2s_shadow_50_text_Q1NETiBA566A5Y2V6ZqP6aOO_size_20_color_FFFFFF_t_70_g_se_x_16 2]: /images/20220829/9c122b01310c4048a697bd3e93102eeb.png