Shuffle-我能找到的最详细的文档-(转)

MapReduce:详解Shuffle过程 博客分类： mapreduce
MapreduceITeye数据结构多线程Hadoop .

        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执行的影响。

        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的方方面面。至于具体的实现原理，各位有兴趣就自己去探索，如果不方便的话，留言给我，我再来研究并反馈。
大小: 62.2 KB
大小: 25.5 KB
大小: 209.5 KB
查看图片附件
10
顶0
踩.分享到：   .[Java拾遗]迭代list过程中删除元素 | MapReduce:Fair Scheduler PPT分享 2011-04-08 14:34浏览 5066评论(51)收藏分类:编程语言相关推荐评论
51 楼 langyu 23 小时前   引用
usezhou 写道
楼主，从什么地方认为 Partition 是在写入内存之前就进行的呢？...

是从0.21版本的Hadoop源码看出这些的。。。

50 楼 usezhou 昨天   引用
楼主，从什么地方认为 Partition 是在写入内存之前就进行的呢？从我的理解，是从内存spill 进硬盘的时候会做分区操作，而不是map输出的时候就做分区。
第二 在spill 进硬盘之前做 的combiner 操作，请问从什么地方看出来，是在此步骤做的combiner 操作呢，我认为combiner 操作作为一个local reducer 应该是在 map部分结束的时候就开始做copy 操作 而没有做完分区的数据 是不可能流向combiner阶段的，所以
我认为：官方给的图是没有错误的，在内存spill 的过程先做分区Partition -->然后sort-->combiner------->reduce
49 楼 langyu 2012-09-06   引用
ddvk2007 写道
。。。关于blobstore。。。

观察你所发出的关于blobstore的链接。。。
Google所描述的blobstore就是一块存储，没有特定类型，可以是本地磁盘，也可以是HDFS。

Google也没有明确表示它的blobstore是哪种类型，大体来说，应该是本地磁盘，Hadoop也是选择本地磁盘作为blobstore。

这篇blog主要说明的就是Shuffle的整个过程，后面半部分都有说到Reducer抓取Mapper存储在本地磁盘的整个流程。

如果有其它不明白的，请联系我：[e-mail]
48 楼 langyu 2012-09-06   引用
ddvk2007 写道
樓主你好
請問
在经过mapper的运行后，我们得知mapper的输出是这样一个key/value对： key是“aaa”， value是数值1。因为当前map端只做加1的操作，在reduce task里才去合并结果集。前面我们知道这个job有3个reduce task，到底当前的“aaa”应该交由哪个reduce去做呢，是需要现在决定的。

為什麼會有三個reduce task? 這部分我不太了解 可能請樓主在說明一下 謝謝
還有 map 與 mapper 一樣嗎? 還是它們有差別?


Reduce task的数量是由我们自己设定的，我当时就设置了3，在Job中有这样的设置方法。
Mapper是运行Map task的那个执行器，你可以把它们看成等同。Map task是逻辑上的概念，一般在说到Job时提到。Mapper是执行时期的概念，一般在说到Shuffle时提到。
47 楼 ddvk2007 2012-09-05   引用
樓主你好
我最近正在研究關於google mapreduce的研究
google 的程序的確也是 map→shuffle→reduce

mapreduce以WordCount為例
map：做讀取字串的動作
shuffle：將相同的字串做+1的動作
reduce：整合結果並以txt檔輸出結果

在每個工作執行的過程中也都會將資訊暫存到一個區塊
而在工作結束後就會將資料儲存於blobstore中
這裡的blobstore可以將它想像成當工作結束後，必須要把結果儲存到其中，
這樣才可以攻下一個工作程序抓取資料
例如：map工作結束後將檔案儲存至blobstore，當shuffle開始運行時，就會到blobstore抓取資料。
關於blobstore的資訊可以參考這裡
https://developers.google.com/appengine/docs/python/blobstore/overview?hl=zh-tw

然後我想要請問的是shuffle的抓取機制是怎麼設計的?
不知道hadoop跟google的機制是否相同
麻煩樓主幫忙了...謝謝


46 楼 ddvk2007 2012-09-05   引用
樓主你好
請問
在经过mapper的运行后，我们得知mapper的输出是这样一个key/value对： key是“aaa”， value是数值1。因为当前map端只做加1的操作，在reduce task里才去合并结果集。前面我们知道这个job有3个reduce task，到底当前的“aaa”应该交由哪个reduce去做呢，是需要现在决定的。

為什麼會有三個reduce task? 這部分我不太了解 可能請樓主在說明一下 謝謝
還有 map 與 mapper 一樣嗎? 還是它們有差別?
45 楼 langyu 2012-09-03   引用
ddvk2007 写道
樓主我想請問 mapreduce 要如何跟pso作結合?
如果樓主有相關的資料是本可以提供給我參考?
google的mapreduce 或是 hadoop的都可以
麻煩樓主了

抱歉，没有相关的资料
Google下，发现这篇文章，不知道是不是你所需要的：http://www.cs.york.ac.uk/rts/docs/CEC-2007/html/pdf/1802.pdf
44 楼 ddvk2007 2012-09-01   引用
樓主我想請問 mapreduce 要如何跟pso作結合?
如果樓主有相關的資料是本可以提供給我參考?
google的mapreduce 或是 hadoop的都可以
麻煩樓主了
43 楼 langyu 2012-08-12   引用
ddvk2007 写道
樓主 抱歉我剛剛說錯了...
我想問的是google 的 mapreduce中
在每個工作程序結束時 會先將資料存到blobstore 以便讓下一個工作程序抓取
而在shuffle 與 reduce 為什麼可以一直從blobstore抓取資料?(我不知道那行程式碼在哪?)
麻煩樓主 謝謝



联系前后所述，我猜你这里的blobstore，应该指的是HDFS。如果两个Job有依赖关系，那么它们之前的确是通过HDFS来共享数据。
但是，Shuffle中的Reduce数据输入不是来自于HDFS，它是来自于Map端写到本地磁盘中的文件。Reduce的输出才会写到HDFS去。
Shuffle过程中，只有Map才源源不断地从HDFS读取数据做运算。
如果你想获取代码的话，那么请参照源码MapTask与ReduceTask这两个类
42 楼 ddvk2007 2012-08-12   引用
樓主 抱歉我剛剛說錯了...
我想問的是google 的 mapreduce中
在每個工作程序結束時 會先將資料存到blobstore 以便讓下一個工作程序抓取
而在shuffle 與 reduce 為什麼可以一直從blobstore抓取資料?(我不知道那行程式碼在哪?)
麻煩樓主 謝謝


41 楼 langyu 2012-08-12   引用
ddvk2007 写道
樓主 我想請問關於google 的mapreduce
為什麼他shuffle 為什麼會重複運行(我在程式碼中並沒有看到用來重複呼叫迴圈的語法)?
謝謝


Google MapReduce的Shuffle会重复运行吗？抱歉，论文中我没有看到这样的描述。是的，那个Shuffle在每个Job中是一次性的，样例代码中也没有重复执行的语法。
如果你再有疑问的话，麻烦附带参考资料，我们一同分析。
40 楼 ddvk2007 2012-08-12   引用
樓主 我想請問關於google 的mapreduce
為什麼他shuffle 為什麼會重複運行(我在程式碼中並沒有看到用來重複呼叫迴圈的語法)?
謝謝


39 楼 langyu 2012-08-10   引用
ddvk2007 写道
我想請問google的mapreduce
他的程序是 map→shuffle→reduce
那要麼樣才可以讓reduce回到shuffle進行疊代呢?

重新启动一个Job

38 楼 ddvk2007 2012-08-10   引用
我想請問google的mapreduce
他的程序是 map→shuffle→reduce
那要麼樣才可以讓reduce回到shuffle進行疊代呢?
37 楼 langyu 2012-05-21   引用
ansjsun 写道

如果最终的多个reducer没有合并.那么.
在对于排序这种需求..只能在Combiner 中不是通过hash来分配reducer..通过大小来区分..可以
或者再用reducer的结果..再跑一次..job  ...来合并为一个....


对，这种需求，自己扩展。要么再来一次Job，要么在Partition上下功夫

36 楼 ansjsun 2012-05-21   引用
langyu 写道
ansjsun 写道

reduce是多个的情况下...加入以排序为例...
那么单独一个reduce..处理完毕后..的操作写到硬盘或这驻留内存中
但是最终结果的处理应该是在一个reduce上吧...
那么多个reduce也是一个归并的过程???这个细节是怎么处理的

这个问题你可以试下，如果有多个Reducer的话，会出现什么情况。
首先，Reducer的输出数据是直接写入HDFS的。其实，如果有多个Reducer的话，那么在HDFS的output path下就会有多个像这样的“part-xxx”文件。文件是以Reducer number为标识，每个文件对应着一个Reducer。所以没有最终多个Reducer的归并过程。




如果最终的多个reducer没有合并.那么.
在对于排序这种需求..只能在Combiner 中不是通过hash来分配reducer..通过大小来区分..可以

或者再用reducer的结果..再跑一次..job  ...来合并为一个....


35 楼 langyu 2012-05-21   引用
ansjsun 写道

reduce是多个的情况下...加入以排序为例...
那么单独一个reduce..处理完毕后..的操作写到硬盘或这驻留内存中
但是最终结果的处理应该是在一个reduce上吧...
那么多个reduce也是一个归并的过程???这个细节是怎么处理的

这个问题你可以试下，如果有多个Reducer的话，会出现什么情况。
首先，Reducer的输出数据是直接写入HDFS的。其实，如果有多个Reducer的话，那么在HDFS的output path下就会有多个像这样的“part-xxx”文件。文件是以Reducer number为标识，每个文件对应着一个Reducer。所以没有最终多个Reducer的归并过程。

34 楼 ansjsun 2012-05-21   引用
感谢楼主的无私奉献.写的很好..我现在有一个疑惑是..
reduce是多个的情况下...加入以排序为例...

那么单独一个reduce..处理完毕后..的操作写到硬盘或这驻留内存中


但是最终结果的处理应该是在一个reduce上吧...


那么多个reduce也是一个归并的过程???这个细节是怎么处理的
33 楼 329937021 2012-05-10   引用
langyu 写道
329937021 写道
这里问个问题，reduce进程是在什么时候启动的呢？还有就是每个reducer到map那边取属于自己的那块数据的时候，是取map结果所有数据呢？还是取map结果中属于这个reducer的数据？能否分析下代码？感谢！

reduce task是当正在运行+完成的task到达一定比例就由JobTracker分配运行。Reducer到map端取数据时当然只取partition到它自己的那一部分数据。如果有两个reduce task，那么第一个reducer task的index就是0，到map端取数据时也是取map端所有对key partition后，属于reduce 0的那些数据

楼主，但是我在源码中没有看到这块代码，MapOutputCopier类中的copyOutput()那一块是写清楚了reducer是取属于自己的那块数据的呢？能否加个qq聊一下，私信给你吧。
32 楼 langyu 2012-04-24   引用
lookqlp 写道

你看看那篇文章，里边有个结论，一个split不可能有零点几个block，也就是说一个split对应多个block。你看看他分析的对不对

好吧，我在文章中声明我分析是基于Hadoop 0.21版本，他说他基于0.18.3。在0.20版本以后，client不能设置你想要几个map task，由jobclient来自动分配，一个split可能就是零点几个HDFS block。