Spark性能优化二 Shuffle机制分析

（一） 什么情况下发生shuffle

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

在Spark中，什么情况下，会发生shuffle？
reduceByKey、groupByKey、sortByKey、countByKey、join等操作都会产生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存在一些不足所以就把它替换掉了。

（二）未优化的Hash Based Shuffle

假设我们是在执行一个reduceByKey之类的操作，此时就会产生shuffle。
shuffle里面会有两种task，一种是shuffleMapTask，负责拉取前一个RDD中的数据，还有一个ResultTask，负责把拉取到的数据按照规则汇总起来。

1：假设有1个节点，这个节点上有2个CPU，上面运行了4个ShuffleMapTask，这样的话其实同时只有2个ShuffleMapTask是并行执行的，因为一个cpu core同时只能执行一个ShuffleMapTask。
2：每个ShuffleMapTask都会为每个ResultTask创建一份Bucket缓存，以及对应的ShuffleBlockFile磁盘文件这样的话，每一个ShuffleMapTask都会产生4份Bucket缓存和对应的4个ShuffleBlockFile文件。
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

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

为了让 Spark 能在更大规模的集群上高性能处理大规模的数据，因此 Spark 引入了 Sort-Based Shuffle。

该机制针对每一个 ShuffleMapTask 都只创建一个文件，将所有的 ShuffleMapTask 的数据都写入同一个文件，并且对应生成一个索引文件。
以前的数据是放在内存中，等到数据写完了再刷写到磁盘，现在为了减少内存的使用，在内存不够用的时候，可以将内存中的数据溢写到磁盘，结束的时候，再将这些溢写的文件联合内存中的数据一起进行归并，从而减少内存的使用量。一方面文件数量显著减少，另一方面减少缓存所占用的内存大小，而且同时避免 GC 的风险和频率。