hadoop2 作业执行过程之reduce过程

---

reduce阶段就是处理map的输出数据，大部分过程和map差不多

---

//ReduceTask.run方法开始和MapTask类似，包括initialize()初始化，根据情况看是否调用runJobCleanupTask()，
  //runJobSetupTask()，runTaskCleanupTask()。之后进入正式的工作，主要有这么三个步骤：Copy、Sort、Reduce。
  @Override
  @SuppressWarnings("unchecked")
  public void run(JobConf job, final TaskUmbilicalProtocol umbilical)
    throws IOException, InterruptedException, ClassNotFoundException {
    this.umbilical = umbilical;
    job.setBoolean("mapred.skip.on", isSkipping());
    /*添加reduce过程需要经过的几个阶段。以便通知TaskTracker目前运   行的情况*/ 
    if (isMapOrReduce()) {
      copyPhase = getProgress().addPhase("copy");
      sortPhase  = getProgress().addPhase("sort");
      reducePhase = getProgress().addPhase("reduce");
    }
    // start thread that will handle communication with parent
 // 设置并启动reporter进程以便和TaskTracker进行交流 
    TaskReporter reporter = new TaskReporter(getProgress(), umbilical,
        jvmContext);
    reporter.startCommunicationThread();
    //在job client中初始化job时，默认就是用新的API，详见Job.setUseNewAPI()方法
    boolean useNewApi = job.getUseNewReducer();        
    /*用来初始化任务，主要是进行一些和任务输出相关的设置，比如创建commiter，设置工作目录等*/ 
    initialize(job, getJobID(), reporter, useNewApi);//这里将会处理输出目录
    /*以下4个if语句均是根据任务类型的不同进行相应的操作，这些方 法均是Task类的方法，所以与任务是MapTask还是ReduceTask无关*/ 
    // check if it is a cleanupJobTask
    if (jobCleanup) {
      runJobCleanupTask(umbilical, reporter);
      return;
    }
    if (jobSetup) {
        //主要是创建工作目录的FileSystem对象
      runJobSetupTask(umbilical, reporter);
      return;
    }
    if (taskCleanup) {
         //设置任务目前所处的阶段为结束阶段，并且删除工作目录 
      runTaskCleanupTask(umbilical, reporter);
      return;
    }
    
    // Initialize the codec
    codec = initCodec();

    boolean isLocal = "local".equals(job.get("mapred.job.tracker", "local"));　　//判断是否是单机hadoop
    if (!isLocal) {
        //1. Copy.就是从执行各个Map任务的服务器那里，收到map的输出文件。拷贝的任务，是由ReduceTask.ReduceCopier 类来负责。
        //ReduceCopier对象负责将Map函数的输出拷贝至Reduce所在机器 
      reduceCopier = new ReduceCopier(umbilical, job, reporter);
      if (!reduceCopier.fetchOutputs()) {////fetchOutputs函数负责拷贝各个Map函数的输出 
        if(reduceCopier.mergeThrowable instanceof FSError) {
          throw (FSError)reduceCopier.mergeThrowable;
        }
        throw new IOException("Task: " + getTaskID() + 
            " - The reduce copier failed", reduceCopier.mergeThrowable);
      }
    }
    copyPhase.complete();                         // copy is already complete
    setPhase(TaskStatus.Phase.SORT);
    statusUpdate(umbilical);

    final FileSystem rfs = FileSystem.getLocal(job).getRaw();
    //2.Sort(其实相当于合并).排序工作，就相当于上述排序工作的一个延续。它会在所有的文件都拷贝完毕后进行。
    //使用工具类Merger归并所有的文件。经过这一个流程，一个合并了所有所需Map任务输出文件的新文件产生了。
    //而那些从其他各个服务器网罗过来的 Map任务输出文件，全部删除了。
    
    //根据hadoop是否分布式来决定调用哪种排序方式 
    RawKeyValueIterator rIter = isLocal
      ? Merger.merge(job, rfs, job.getMapOutputKeyClass(),
          job.getMapOutputValueClass(), codec, getMapFiles(rfs, true),
          !conf.getKeepFailedTaskFiles(), job.getInt("io.sort.factor", 100),
          new Path(getTaskID().toString()), job.getOutputKeyComparator(),
          reporter, spilledRecordsCounter, null)
      : reduceCopier.createKVIterator(job, rfs, reporter);
        
    // free up the data structures
    mapOutputFilesOnDisk.clear();
    
    sortPhase.complete();                         // sort is complete
    setPhase(TaskStatus.Phase.REDUCE); 
    statusUpdate(umbilical);
    //3.Reduce 1.Reduce任务的最后一个阶段。它会准备好Map的 keyClass（"mapred.output.key.class"或"mapred.mapoutput.key.class"）, 
    //valueClass("mapred.mapoutput.value.class"或"mapred.output.value.class")
    //和 Comparator （“mapred.output.value.groupfn.class”或 “mapred.output.key.comparator.class”）
    Class keyClass = job.getMapOutputKeyClass();
    Class valueClass = job.getMapOutputValueClass();
    RawComparator comparator = job.getOutputValueGroupingComparator();
    //2.根据参数useNewAPI判断执行runNewReduce还是runOldReduce。分析润runNewReduce
    if (useNewApi) {
        //3.runNewReducer
        //0.像报告进程书写一些信息
        //1.获得一个TaskAttemptContext对象。通过这个对象创建reduce、output及用于跟踪的统计output的RecordWrit、最后创建用于收集reduce结果的Context
        //2.reducer.run(reducerContext)开始执行reduce
      runNewReducer(job, umbilical, reporter, rIter, comparator, 
                    keyClass, valueClass);
    } else {
      runOldReducer(job, umbilical, reporter, rIter, comparator, 
                    keyClass, valueClass);
    }
    done(umbilical, reporter);
  }

1.reduce过程中三个大的阶段比较重要：Copy、Sort、Reduce；

2.codec = initCodec()这句是检查map的输出是否是压缩的，压缩的则返回压缩codec实例，否则返回null，这里讨论不压缩的；

3.实际中使用完全分布式的hadoop，即isLocal==false，然后构造一个ReduceCopier对象reduceCopier，并调用reduceCopier.fetchOutputs()方法拷贝各个Mapper的输出，到本地；

4.done(umbilical, reporter)这个方法用于做结束任务的一些清理工作：更新计数器updateCounters()；如果任务需要提交，设置Taks状态为COMMIT_PENDING，并利用TaskUmbilicalProtocol，汇报Task完成，等待提交，然后调用commit提交任务；设置任务结束标志位；结束Reporter通信线程；发送最后一次统计报告(通过sendLastUpdate方法)；利用TaskUmbilicalProtocol报告结束状态（通过sendDone方法)。

---