MapReduce:详解Shuffle过程
Shuffle过程是MapReduce的核心，也被称为奇迹发生的地方。要想理解MapReduce， Shuffle是必需要了解的。我看过不少相关的资料，但每次看完都云里雾里的绕着，很难理清大体的逻辑，反而越搅越混。前段时间在作MapReduce job 性能调优的工做，须要深刻代码研究MapReduce的运行机制，这才对Shuffle探了个究竟。考虑到以前我在看相关资料而看不懂时很恼火，因此在这 里我尽最大的可能试着把Shuffle说清楚，让每一位想了解它原理的朋友都能有所收获。若是你对这篇文章有任何疑问或建议请留言到后面，谢谢！
Shuffle的正常意思是洗牌或弄乱，可能你们更熟悉的是Java API里的Collections.shuffle(List)方法，它会随机地打乱参数list里的元素顺序。若是你不知道MapReduce里 Shuffle是什么，那么请看这张图：
这张是官方对Shuffle过程的描述。但我能够确定的 是，单从这张图你基本不可能明白Shuffle的过程，由于它与事实相差挺多，细节也是错乱的。后面我会具体描述Shuffle的事实状况，因此这里你只 要清楚Shuffle的大体范围就成－怎样把map task的输出结果有效地传送到reduce端。也能够这样理解， Shuffle描述着数据从map task输出到reduce task输入的这段过程。
在Hadoop这样的集群环境中，大部分map task与reduce task的执行是在不一样的节点上。固然不少状况下Reduce执行时须要跨节点去拉取其它节点上的map task结果。若是集群正在运行的job有不少，那么task的正常执行对集群内部的网络资源消耗会很严重。这种网络消耗是正常的，咱们不能限制，能作的 就是最大化地减小没必要要的消耗。还有在节点内，相比于内存，磁盘IO对job完成时间的影响也是可观的。从最基本的要求来讲，咱们对Shuffle过程的 指望能够有：
数组
完整地从map task端拉取数据到reduce 端。性能优化
在跨节点拉取数据时，尽量地减小对带宽的没必要要消耗。网络
减小磁盘IO对task执行的影响。app
OK，看到这里时，你们能够先停下来想一想，若是是本身来设计这段Shuffle过程，那么你的设计目标是什么。我想能优化的地方主要在于减小拉取数据的量及尽可能使用内存而不是磁盘。
个人分析是基于Hadoop0.21.0的源码，若是与你所认识的Shuffle过程有差异，不吝指出。我会以WordCount为例，并假设它有8个 map task和3个reduce task。从上图看出，Shuffle过程横跨map与reduce两端，因此下面我也会分两部分来展开。
先看看map端的状况，以下图：
上图多是某个map task的运行状况。拿它与官方图的左半边比较，会发现不少不一致。官方图没有清楚地说明partition， sort与combiner到底做用在哪一个阶段。我画了这张图，但愿让你们清晰地了解从map数据输入到map端全部数据准备好的全过程。
整个流程我分了四步。简单些能够这样说，每一个map task都有一个内存缓冲区，存储着map的输出结果，当缓冲区快满的时候须要将缓冲区的数据以一个临时文件的方式存放到磁盘，当整个map task结束后再对磁盘中这个map task产生的全部临时文件作合并，生成最终的正式输出文件，而后等待reduce task来拉数据。
固然这里的每一步均可能包含着多个步骤与细节，下面我对细节来一一说明：
1. 在map task执行时，它的输入数据来源于HDFS的block，固然在MapReduce概念中，map task只读取split。Split与block的对应关系多是多对一，默认是一对一。在WordCount例子里，假设map的输入数据都是像 “aaa”这样的字符串。
2. 在通过mapper的运行后，咱们得知mapper的输出是这样一个key/value对： key是“aaa”， value是数值1。由于当前map端只作加1的操做，在reduce task里才去合并结果集。前面咱们知道这个job有3个reduce task，到底当前的“aaa”应该交由哪一个reduce去作呢，是须要如今决定的。
MapReduce提供Partitioner接口，它的做用就是根据key或value及reduce的数量来决定当前的这对输出数据最终应该交由哪一个 reduce task处理。默认对key hash后再以reduce task数量取模。默认的取模方式只是为了平均reduce的处理能力，若是用户本身对Partitioner有需求，能够订制并设置到job上。
在咱们的例子中，“aaa”通过Partitioner后返回0，也就是这对值应当交由第一个reducer来处理。接下来，须要将数据写入内存缓冲区 中，缓冲区的做用是批量收集map结果，减小磁盘IO的影响。咱们的key/value对以及Partition的结果都会被写入缓冲区。固然写入之 前，key与value值都会被序列化成字节数组。
整个内存缓冲区就是一个字节数组，它的字节索引及key/value存储结构我没有研究过。若是有朋友对它有研究，那么请大体描述下它的细节吧。
3. 这个内存缓冲区是有大小限制的，默认是100MB。当map task的输出结果不少时，就可能会撑爆内存，因此须要在必定条件下将缓冲区中的数据临时写入磁盘，而后从新利用这块缓冲区。这个从内存往磁盘写数据的过 程被称为Spill，中文可译为溢写，字面意思很直观。这个溢写是由单独线程来完成，不影响往缓冲区写map结果的线程。溢写线程启动时不该该阻止map 的结果输出，因此整个缓冲区有个溢写的比例spill.percent。这个比例默认是0.8，也就是当缓冲区的数据已经达到阈值（buffer size * spill percent = 100MB * 0.8 = 80MB），溢写线程启动，锁定这80MB的内存，执行溢写过程。Map task的输出结果还能够往剩下的20MB内存中写，互不影响。
当溢写线程启动后，须要对这80MB空间内的key作排序(Sort)。排序是MapReduce模型默认的行为，这里的排序也是对序列化的字节作的排序。
在这里咱们能够想一想，由于map task的输出是须要发送到不一样的reduce端去，而内存缓冲区没有对将发送到相同reduce端的数据作合并，那么这种合并应该是体现是磁盘文件中 的。从官方图上也能够看到写到磁盘中的溢写文件是对不一样的reduce端的数值作过合并。因此溢写过程一个很重要的细节在于，若是有不少个 key/value对须要发送到某个reduce端去，那么须要将这些key/value值拼接到一块，减小与partition相关的索引记录。
在针对每一个reduce端而合并数据时，有些数据可能像这样：“aaa”/1， “aaa”/1。对于WordCount例子，就是简单地统计单词出现的次数，若是在同一个map task的结果中有不少个像“aaa”同样出现屡次的key，咱们就应该把它们的值合并到一块，这个过程叫reduce也叫combine。但 MapReduce的术语中，reduce只指reduce端执行从多个map task取数据作计算的过程。除reduce外，非正式地合并数据只能算作combine了。其实你们知道的，MapReduce中将Combiner等 同于Reducer。
若是client设置过Combiner，那么如今就是使用Combiner的时候了。将有相同key的key/value对的value加起来，减小溢 写到磁盘的数据量。Combiner会优化MapReduce的中间结果，因此它在整个模型中会屡次使用。那哪些场景才能使用Combiner呢？从这里 分析，Combiner的输出是Reducer的输入，Combiner毫不能改变最终的计算结果。因此从个人想法来看，Combiner只应该用于那种 Reduce的输入key/value与输出key/value类型彻底一致，且不影响最终结果的场景。好比累加，最大值等。Combiner的使用必定 得慎重，若是用好，它对job执行效率有帮助，反之会影响reduce的最终结果。
4. 每次溢写会在磁盘上生成一个溢写文件，若是map的输出结果然的很大，有屡次这样的溢写发生，磁盘上相应的就会有多个溢写文件存在。当map task真正完成时，内存缓冲区中的数据也所有溢写到磁盘中造成一个溢写文件。最终磁盘中会至少有一个这样的溢写文件存在(若是map的输出结果不多，当 map执行完成时，只会产生一个溢写文件)，由于最终的文件只有一个，因此须要将这些溢写文件归并到一块儿，这个过程就叫作Merge。Merge是怎样 的？如前面的例子，“aaa”从某个map task读取过来时值是5，从另一个map 读取时值是8，由于它们有相同的key，因此得merge成group。什么是group。对于“aaa”就是像这样的：{“aaa”, [5, 8, 2, …]}，数组中的值就是从不一样溢写文件中读取出来的，而后再把这些值加起来。请注意，由于merge是将多个溢写文件合并到一个文件，因此可能也有相同的 key存在，在这个过程当中若是client设置过Combiner，也会使用Combiner来合并相同的key。
至此，map端的全部工做都已结束，最终生成的这个文件也存放在TaskTracker够得着的某个本地目录内。每一个reduce task不断地经过RPC从JobTracker那里获取map task是否完成的信息，若是reduce task获得通知，获知某台TaskTracker上的map task执行完成，Shuffle的后半段过程开始启动。
简单地说，reduce task在执行以前的工做就是不断地拉取当前job里每一个map task的最终结果，而后对从不一样地方拉取过来的数据不断地作merge，也最终造成一个文件做为reduce task的输入文件。见下图：
如map 端的细节图，Shuffle在reduce端的过程也能用图上标明的三点来归纳。当前reduce copy数据的前提是它要从JobTracker得到有哪些map task已执行结束，这段过程不表，有兴趣的朋友能够关注下。Reducer真正运行以前，全部的时间都是在拉取数据，作merge，且不断重复地在作。 如前面的方式同样，下面我也分段地描述reduce 端的Shuffle细节：
1. Copy过程，简单地拉取数据。Reduce进程启动一些数据copy线程(Fetcher)，经过HTTP方式请求map task所在的TaskTracker获取map task的输出文件。由于map task早已结束，这些文件就归TaskTracker管理在本地磁盘中。
2. Merge阶段。这里的merge如map端的merge动做，只是数组中存放的是不一样map端copy来的数值。Copy过来的数据会先放入内存缓冲区 中，这里的缓冲区大小要比map端的更为灵活，它基于JVM的heap size设置，由于Shuffle阶段Reducer不运行，因此应该把绝大部分的内存都给Shuffle用。这里须要强调的是，merge有三种形 式：1)内存到内存 2)内存到磁盘 3)磁盘到磁盘。默认状况下第一种形式不启用，让人比较困惑，是吧。当内存中的数据量到达必定阈值，就启动内存到磁盘的merge。与map 端相似，这也是溢写的过程，这个过程当中若是你设置有Combiner，也是会启用的，而后在磁盘中生成了众多的溢写文件。第二种merge方式一直在运 行，直到没有map端的数据时才结束，而后启动第三种磁盘到磁盘的merge方式生成最终的那个文件。
3. Reducer的输入文件。不断地merge后，最后会生成一个“最终文件”。为何加引号？由于这个文件可能存在于磁盘上，也可能存在于内存中。对咱们 来讲，固然但愿它存放于内存中，直接做为Reducer的输入，但默认状况下，这个文件是存放于磁盘中的。至于怎样才能让这个文件出如今内存中，以后的性能优化篇我再说。当Reducer的输入文件已定，整个Shuffle才最终结束。而后就是Reducer执行，把结果放到HDFS上。 上面就是整个Shuffle的过程。细节不少，我不少都略过了，只试着把要点说明白。固然，我可能也有理解或表述上的不少问题，不吝指点。我但愿不断地完 善和修改这篇文章，能让它通俗、易懂，看完就能知道Shuffle的方方面面。至于具体的实现原理，各位有兴趣就本身去探索，若是不方便的话，留言给我， 我再来研究并反馈。