Spark的shuffle机制

最近在面试大数据开发工程师，面某B公司的时候问到了Spark的shuffle机制，并且问和MR的shuffle有什么样的区别，当时答得不太好，决定好好研究这个玩意儿，网上讲的很多，下面我就把我的理解做一下总结。

什么时候shuffle

在解释shuffle机制之前，首先要搞明白什么时候shuffle，那就得讲讲什么是shuffle了，咱们先把英语直译一下：洗牌。玩过扑克的兄弟肯定都不陌生，或者讲得更通俗点，就是打乱顺序。但是在大数据计算里，好像和这个过程有点不太相同。废话不多说，上官方解释：
——Shuffle描述着数据从map task输出到reduce task输入的这段过程。shuffle是连接Map和Reduce之间的桥梁，Map的输出要用到Reduce中必须经过shuffle这个环节，shuffle的性能高低直接影响了整个程序的性能和吞吐量。因为在分布式情况下，reduce task需要跨节点去拉取其它节点上的map task结果。这一过程将会产生网络资源消耗和内存，磁盘IO的消耗。通常shuffle分为两部分：Map阶段的数据准备和Reduce阶段的数据拷贝处理。一般将在map端的Shuffle称之为Shuffle Write，在Reduce端的Shuffle称之为Shuffle Read。
这个官方解释虽然有点长，但是我们提炼其精华就可以知道，shuffle是发生在map和reduce之间的一个过程，MR的shuffle过程比较简单，我们今天先搞个比较难的，研究一下Spark的shuffle机制。在MR计算时，map过程和reduce过程定义的比较清晰(可能和MR的“八股文”编程方式有点关系？)，但是在Spark中map和reduce过程好像不那么明显了，那么什么时候shuffle呢？我们先来看张图：

我们知道Spark计算过程中DAG Scheduler会根据RDD之间的宽窄依赖关系进行Stage划分，对Stage划分和宽窄依赖有所了解的朋友就会知道，同一个Stage内的RDD之间是窄依赖，它们发生的都是map、union、filter等这种一对一或者多对一的操作，只有在Stage和Stage之间也就是宽依赖之间会进行GroupBy、PartitionBy中一对多或者多对多的操作，所以Shuffle过程就发生在这个时候。

谁负责shuffle

在Spark中，有一个负责shuffle过程执行、计算和处理的组件ShuffleManager。随着Spark的发展，现在ShuffleManager已经有两种实现方式：HashShuffleManager和SortShuffleManager，从这两种实现方式中，我们不难猜出Spark的shuffle机制也有两种方式：hash shuffle（spark1.2之前）和sort shuffle（spark1.2以后）。

Hash shuffle

我们先来看看Hash shuffle,虽然都说过时不谈，但是我们还是来研究研究，看一看它过时的原因。

• 优化前

首先看一个优化前的hash shuffle，上图：

我们假设图中每个executor中只有一个core，也就是说每个executor同一时间只能执行一个task线程。
我们来分析分析图中shuffle的write阶段和read阶段。
（1）shuffle write阶段
上面大方框中的task我们可以看成一个map类型的task，该task处理的结果不会直接落盘，会先按照key进行分区(也就是说分配一下这条结果应该由哪一个reduce任务处理)，具体过程是对key执行hash算法(应该就是HashParitioner()的工作)，因此会在本地内存缓冲中为每一个reduce任务建立对应的分区(bucket),当本地内存缓冲写满之后，会将内容溢写到磁盘中，每一个分区会溢写到一个本地文件中，所以map任务结束后，会生成mr个本地文件，m指map任务数量，r指reduce任务数量。设想一下如果有100个map任务，50个reduce任务，将会生成5000个本地小文件，而大量小文件对HDFS这样的文件系统是简直是噩梦（5000128M=725G,可怕不，这还没有算法备份）！
（2）shuffle read阶段
当上一个stage的task将处理结果存储到文件系统，后下一个stage的task要对其进行reduce类型的操作，那么需要先把处理结果拉取到本地，这就是shuffle read阶段要进行的工作，由于上一个阶段已经按照reduce分好区，所以reduce任务节点只需要对上游task所在节点将对应的文件拉取到本地即可。实际上shuffle read是一边拉取一遍进行reduce操作的，每一次拉取一个内存缓冲大小的数据，执行完再拉取下一批，直到将所有结果拉取完。
存在的问题：
分析完这两个阶段，我就知道它为什么会被淘汰了，shuffle过程中产生了海量的小文件，导致了大量的磁盘I/O，大大降低了计算性能，另外读写磁盘时的对象都存储在堆内存中，容易造成堆内存不足而频繁发生GC，GC又会导致OOM。

• 优化后

既然存在问题，就需要进行优化，Spark选择了复用buffer 的方式，也就是说不再为每一个map类型的task创建对应reduce任务数量的分区，而是使在同一个executor core中执行的map task线程共用一个内存缓冲，所以有多少core就会创建多少个包含reduce数量分区的内存缓冲，上图：

这样最后产生的本地文件数量就变成了c*r个，c指执行map任务的executor中core的总数量，r指reduce任务数量，core数量通常要比map任务数量少很多，这也就大大减少了产生小文件的数量。虽然这种方式在一定程度上减少了小文件产生的数量，但是当reduce任务过多时，依然不可避免的产生大量小文件。

Sort shuffle

不合理的方式总会被淘汰，既然hash shuffle存在不合理的地方，那么终究逃脱不了被替代的命运，取而代之的就是sort shuffle，从名字上可以看出来这是一种基于排序的shuffle，那么到底是不是呢？深入研究一下。
实际上sort shuffle也有两种运行机制，一种是普通运行机制，另一种是bypass运行机制。

普通运行机制

首先来看一下普通运行机制，上图：

从图中可以看出，sort shuffle和hash shuffle相比确实发生了很大的变化，很明显的一点就是不会再根据下游任务数量建立对应分区了，详细看一下，分为这几个过程：
(1)写入内存数据结构
上游任务产生的结果不再分区写了，而是全部写入一个内存数据结构中(默认是5M)，只不过用的数据结构可能有些不同，对于reduceByKey这种算子，用的是Map这种<k,v>型数据结构，而对于join这种算子，就用的是Array数组这种数据结构。
这里有个需要注意的点，每一次写入一条数据之后，就会进行判断一下是否超过某个阈值，如果超过就会把内存数据结构中的数据溢写到磁盘，然后把内存数据结构清空。但是这个阈值是动态变化的，shuffle中有个定时器会定时检查内存数据结构大小是否够用，如果不够就会申请额外的内存，只有当内存不够申请不到内存时，才会发生溢写。申请内存的数量为：
申请的内存=当前的内存*2-上一次的内存(有点难以理解，但是细扣一下还是能理解的)
(2)排序
但是实际上在溢写到磁盘之前，还会根据key对内存数据结构中的数据进行排序。这就体现出sort shuffle的特点了，然后再思考一下，为什么要排序呢？我觉得跟下面的这个过程有关。
(3)溢写
发生溢写的时候，sort shuffle里没有像hash shuffle里那种分区，但是又不能把所有的数据都写入到一个溢写文件中，所以就要分开写入多个文件，规定每个文件写入多少条数据，那么按照什么规则写呢？来一条写一条？肯定不能这么干，因为这么干到下一个stage拉取数据进行合并的时候估计就疯了。那就按照key排一下序吧，然后把排序后的结果每10000条分批写入磁盘文件中。
思考：看一下排序和溢写这两个过程，是不是和hash shuffle中的分区然后溢写有点像呢？只不过不再按照key的hash值进行分区，而是根据排序后的key进行分段了。
(4)merge
从内存数据结构溢写到磁盘时，产生了多个磁盘文件，咦，那这不就和hash shuffle一样了吗？骚年，既然人家spark推出了新的机制，那肯定就不允许让它一样。事实上，从内存数据结构溢写到磁盘时，产生的都是临时的磁盘文件，最后要将所有的临时磁盘文件进行一次合并，也就是merge，这样最终就只产生了一个磁盘文件，这时又会有一个问题，每一个executor只产生了一个磁盘文件，而下游stage的多个reduce任务都会从这一个文件中拉取数据，每个任务拉哪些数据呢？为了解决一个问题，产生磁盘文件的时候还会为这个文件生成一个索引文件，这个索引文件中就标识出了下游的各个任务从磁盘文件中拉取数据的起止位置start offset和end offset。
我们再来分析分析sort shuffle一共产生了多少磁盘文件：磁盘文件数量=2m
m是map任务的数量，乘2是因为map任务会产生一个数据文件和一个索引文件。
由此来看，sort shuffle确实会比hash shuffle产生的磁盘文件(mr或者c*r)少，因为reduce task的数量通常是要远大于2的。

Bypass运行机制

上面基于排序的sort shuffle确实能够减少产生的磁盘文件数量，但是sort的过程无疑会增加一定的性能开销，那么能不能不排序呢，比如说当下游reduce任务比较少的时候？答案是，能！sort shuffle里设置了一个参数，当shuffle read也就是下游reduce任务数量小于spark.shuffle.sort.bypassMergeThreshold(默认200)时候，就bypass，即绕过sort过程。这时候分区方式就和未优化的hash shuffle比较像了，每一个map任务都会建立对应reduce数量的分区，根据key的hash值将结果写入对应的分区，当然也是先写缓存再溢写，但是，这里有一个但是，它会和普通sort shuffle一样把这些产生的磁盘文件最终合并成一个磁盘文件，同时也创建一个索引文件方便reduce任务拉取文件。不说了，上图：

这个图网上很多图都画成了三个内存缓冲，其实应该是对应下游reduce任务数量的两个内存缓冲并溢写到两个临时磁盘文件中,所以我又重画了一下。
bypass机制虽然和hash shuffle一样中途产生了大量的小文件，比普通sort shuffle要多，但是胜在它数量少，因此在阈值合适的情况下速度也是很快的，毕竟省去了排序那一步。

到这里，spark shuffle机制已经研究个差不多了，当然这只是表面的一些逻辑，要想真正弄明白咋回事，还需要进一步看看源码。
最后引用一下Spark中的Spark Shuffle详解中的一段总结，我觉得写的很清晰：
Shuffle 过程本质上都是将 Map 端获得的数据使用分区器进行划分，并将数据发送给对应的 Reducer 的过程。

shuffle作为处理连接map端和reduce端的枢纽，其shuffle的性能高低直接影响了整个程序的性能和吞吐量。map端的shuffle一般为shuffle的Write阶段，reduce端的shuffle一般为shuffle的read阶段。Hadoop和spark的shuffle在实现上面存在很大的不同，spark的shuffle分为两种实现，分别为HashShuffle和SortShuffle，

HashShuffle又分为普通机制和合并机制，普通机制因为其会产生mr个数的巨量磁盘小文件而产生大量性能低下的I/O操作，从而性能较低，因为其巨量的磁盘小文件还可能导致OOM，HashShuffle的合并机制通过重复利用buffer从而将磁盘小文件的数量降低到cr个，但是当Reducer 端的并行任务或者是数据分片过多的时候，依然会产生大量的磁盘小文件。r

SortShuffle也分为普通机制和bypass机制，普通机制在内存数据结构(默认为5M)完成排序，会产生2m个磁盘小文件。而当shuffle map task数量小于spark.shuffle.sort.bypassMergeThreshold参数的值。或者算子不是聚合类的shuffle算子(比如reduceByKey)的时候会触发SortShuffle的bypass机制，SortShuffle的bypass机制不会进行排序，极大的提高了其性能

在Spark 1.2以前，默认的shuffle计算引擎是HashShuffleManager，因为HashShuffleManager会产生大量的磁盘小文件而性能低下，在Spark 1.2以后的版本中，默认的ShuffleManager改成了SortShuffleManager。SortShuffleManager相较于HashShuffleManager来说，有了一定的改进。主要就在于，每个Task在进行shuffle操作时，虽然也会产生较多的临时磁盘文件，但是最后会将所有的临时文件合并(merge)成一个磁盘文件，因此每个Task就只有一个磁盘文件。在下一个stage的shuffle read task拉取自己的数据时，只要根据索引读取每个磁盘文件中的部分数据即可。

特别感谢：博主 大葱拌豆腐 以及他的分享Spark中的Spark Shuffle详解，如侵必删！