Hadoop 1.x的Shuffle源码分析之1

本文主要是介绍Hadoop 1.x的Shuffle源码分析之1，希望对大家解决编程问题提供一定的参考价值，需要的开发者们随着小编来一起学习吧！

先参考董西成的博文  http://dongxicheng.org/mapreduce/hadoop-shuffle-phase/   

Hadoop中shuffle阶段流程分析

Hadoop的一个任务执行过程，分为Map和Reduce两个阶段。而shuffle发生在Reducer阶段。Hadoop 1.2.1里，Reduce类的源码在org.apache.hadoop.mapreduce包的Reducer.java文件，这里有一份详细的reduce过程的解释。shuffle是reduce的第一个阶段，以http的方式从map任务的输出获取数据。reduce的第二个阶段是sort，根据key对来reducer的输入进行排序，因为不同的map任务可能会产生相同key的输出。reducer的第三个阶段就是做最终处理，根据Reducer的reduce函数处理数据。

Reducer类只是提供了一个可Override的reduce函数，Shuffle实际上在ReduceTask类执行。ReduceTask类在org.apache.hadoop.mapred包里。

ReduceTask类比较复杂，有7个内嵌类，有些内嵌类里也有自己的内嵌类，毕其功于一役的做法。它的主要代码在run函数执行：

这是reduce的三个阶段：

if (isMapOrReduce()) {copyPhase = getProgress().addPhase("copy");sortPhase  = getProgress().addPhase("sort");reducePhase = getProgress().addPhase("reduce");}

shuffle阶段在这里

reduceCopier = new ReduceCopier(umbilical, job, reporter);if (!reduceCopier.fetchOutputs()) {if(reduceCopier.mergeThrowable instanceof FSError) {throw (FSError)reduceCopier.mergeThrowable;}throw new IOException("Task: " + getTaskID() + " - The reduce copier failed", reduceCopier.mergeThrowable);}

fetchOutputs函数接近400行，比较长，略坑，在这个函数，

copiers = new ArrayList<MapOutputCopier>(numCopiers);

创建了一组 MapOutputCopier，它是个线程类，负责取数据。

//start the on-disk-merge threadlocalFSMergerThread = new LocalFSMerger((LocalFileSystem)localFileSys);//start the in memory merger threadinMemFSMergeThread = new InMemFSMergeThread();localFSMergerThread.start();inMemFSMergeThread.start();// start the map events threadgetMapEventsThread = new GetMapEventsThread();getMapEventsThread.start();

再创建几个线程，有的负责取map任务的信息，有的负责对结果做归并。

继续往下，是一个while循环，这个循环处理取数据，有一百多行代码，当数据取完或者达到失败上限就终止循环。

循环结束后，依次终止 获取map事件线程， 取数据线程，shuffl内存管理线程，排序线程，至此shuffle就结束了。

这篇关于Hadoop 1.x的Shuffle源码分析之1的文章就介绍到这儿，希望我们推荐的文章对编程师们有所帮助！

---