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

原创
2017/01/17 09:44
阅读数 60

先参考董西成的博文  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 thread
      localFSMergerThread = new LocalFSMerger((LocalFileSystem)localFileSys);
      //start the in memory merger thread
      inMemFSMergeThread = new InMemFSMergeThread();
      localFSMergerThread.start();
      inMemFSMergeThread.start();
      
      // start the map events thread
      getMapEventsThread = new GetMapEventsThread();
      getMapEventsThread.start();

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


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


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

展开阅读全文
打赏
0
0 收藏
分享
加载中
更多评论
打赏
0 评论
0 收藏
0
分享
返回顶部
顶部