Spark内核分析之Shuffle操作流程（非常重要）

如题，我们来分析一下spark的shuffle操作原理；为什么说其非常重要，是因为shuffle操作是我们在Spark调优中非常重要的一环，对shuffle进行了优化，往往可以使得我们的spark程序运行效率有极大的提升。依照惯例，我们先来看一张图；

普通shuffle流程图

上图是一个普通的Shuffle操作流程原理图，一个shuffle操作由三个RDD算子构成，分别是mapPartitionsRDD，ShuffleRDD，mapPartitionsRDD；如上图所示，

1.每个ShuffleMapTask都会为每个ResultTask创建一个Bucket缓存和一个对应的ShuffleBlockFile磁盘文件；2.每个ShuffleMapTask的输出相关信息封装成一个MapStatus发送到DAGScheduler的MapOutputTracker中去； 3.ResultTask开始拉取该任务所需要的数据，ResultTask通过向DAGScheduler的MapOutputTracker获取MapStatus的信息，从而知道自己需要的数据所在的位置，然后去相应的位置拉去数据到该任务所在节点的内存中，如果内存不够，会将部分数据写入磁盘，完成这系列的操作是由ShufflerRDD算子完成的； 4.然后ResultTask对拉取到的数据进行聚合操作，最后生成mapPartitionsRDD算子；

想想上面的这个Shuffler流程会有什么问题？

我们来做一个假设，如果有100个ShuffleMapTask，2个CPU Core，100个ResultTask，那么这个shuffler操作将产生10000个文件，如此多的文件对于Spark作业的性能就是一个灾难；针对这个问题当然有对应的优化策略，接着我们来看另外一张图；

优化的Shuffler流程图

        通过优化的Shuffler操作如上图所示，假设有100个ShufflerMapTask，2CPU core，100个Resulttask，优化后产生的中间文件是200个，是优化之前的1/50；那么这是如何做到的，通过阅读源码可以知道，只要引入consolidation机制就可以实现了，其配置是通过在SparkConf中配置对应的参数即可实现；

        来简单分析一下：一个ShuffleMapTask将数据写入本地不变，但是当这一批ShuffleMapTask运行完成以后，下一批ShuffleMapTask开始运行（一批ShuffleMapTask是指，同一时间有两个Task并行执行，因为有两个CPU Core），它们产生的数据会直接写入上一批ShuffleMapTask产生的本地文件中；上图中左边的一组可以称为一组ShuffleGroup，每个文件中都存储了多个ShuffleMapTask的数据，每个ShuffleMapTask所产生的数据是一个segment，每个File中通过索引，偏移量来标记每部分数据来自不同的ShuffleMapTask。

下面我们来看看源码是如何实现的；

ShuffleMapTask的runTask

1.首先通过Spark全局变量得到shuffleManager对象，并通过shuffleManager对象获得Write对象；

2.接着，通过rdd.iterator方法对属于自己的partition进行计算，最后会调用我们自己编写的RDD算子来计算partition；

3.接着Writer调用自己的write方法将RDD算子计算的结果写入缓存；

HashShuffleWriter的write

1.判断aggregator为true，并且是否设置了map端的combine操作；若成立，则进行map端的数据合并（这里是一个spark优化点，在我之前关于spark优化系列文章中有写过）；

2.对所有经过合并操作之后的数据遍历，根据每个元素获得对应的bucketId，然后将改元素写入对应的bucket缓存中；

这里我们来看看这个shuffle对象做了什么？

FileShuffleBlockManager的forMapTask

1.首先创建出一个ShuffleWriterGroup对象；

2.接着判断Spark作业是否设置了consolidateShuffleFiles；如果设置其为true，首先得到一个fileGroup对象，然后使用shuffleId，mapId，BucketId来得到一个blockId，接着根据这个blockId写数据到磁盘的对象；相反，如果没有设置consolidateShuffleFiles为true，则直接为每个shuffleMapTask创建一个blockFile，然后得到一个写数据到磁盘的对象；

3.执行完这里后，接着调用write方法将数据写入内存缓冲bucket，然后再将数据写入磁盘；

写数据到这里就完成了，然后会将产生的数据位置等信息封装成一个MapStatus对象发送给DAGSchedule的MapOutputTracker中；接下来ResultTask开始读取数据；

ShuffleRDD的compute

HashShuffleReader的read

BlockStoreShuffleFetcher的fetch

BlockStoreShuffleFetcher的fetch

总结：到此shuffle的整个操作流程就分析完了，接下来会分析底层数据存储的核心组件BlockManager的工作原理，，欢迎关注。

如需转载，请注明：

本篇：Spark内核分析之Shuffle操作流程（非常重要）