MapTask（Hadoop2.7.3）

MapTask.java繼承於Task，是hadoop中Map節點主要所做的主要流程。
一般被jvmtask初始化或者在MapTaskAttemptImpl被初始化。其主要流程寫在run()方法中。
run()方法主要在YarnChild中被呼叫（偽分散式或者分散式，如果是在單機模式一般被LocalJobRunning被調）也就是說Map節點都是被Yarn啟動的。

@Override
  public void run(final JobConf job, final TaskUmbilicalProtocol umbilical)
          throws
 
 IOException, ClassNotFoundException, InterruptedException {
    this.umbilical = umbilical;

    if (isMapTask()) {
      // If there are no reducers then there won't be any sort. Hence the map
      // phase will govern the entire attempt's progress.
      if (conf.getNumReduceTasks() == 0) {
        mapPhase = getProgress().addPhase("map"
 
, 1.0f);
      } else {
        // If there are reducers then the entire attempt's progress will be
        // split between the map phase (67%) and the sort phase (33%).
        mapPhase = getProgress().addPhase("map", 0.667f);
        sortPhase = getProgress().addPhase("sort", 0.333f);
      }
    }
    TaskReporter reporter = startReporter(umbilical);

    boolean
 
 useNewApi = job.getUseNewMapper();
    initialize(job, getJobID(), reporter, useNewApi);

    // check if it is a cleanupJobTask
    if (jobCleanup) {
      runJobCleanupTask(umbilical, reporter);
      return;
    }
    if (jobSetup) {
      runJobSetupTask(umbilical, reporter);
      return;
    }
    if (taskCleanup) {
      runTaskCleanupTask(umbilical, reporter);
      return;
    }
    //sendSend(umbilical);
    if (useNewApi) {
      System.out.println("新事前："+umbilical.getClass().getName());
      LOG.info("新事前："+umbilical.getClass().getName());
      runNewMapper(job, splitMetaInfo, umbilical, reporter);
    } else {
      System.out.println("舊事前："+umbilical.getClass().getName());
      LOG.info("舊事前："+umbilical.getClass().getName());
      runOldMapper(job, splitMetaInfo, umbilical, reporter);
    }
    LOG.info("事後："+umbilical.getClass().getName());
    System.out.println("事後："+umbilical.getClass().getName());
    //sendSend(umbilical);
    done(umbilical, reporter);

  }

這部分是我加過log的run()方法，因為圖省事，找日誌找起來比較方便，就直接使用System.out.println()來做輸出了。這樣的話當執行完之後，所有的輸出都會在hadoop目錄下logs/userlogs/stdout裡面。

可以看到run()方法主要有兩個引數：

• final JobConf job
  第一個引數主要是 Job的一些配置，可能會影響到maptask執行時候的一些方式。

• final TaskUmbilicalProtocol umbilical
  這個引數就比較重要了，主要是的事件通訊都是通過這個物件的。具體的在說MapTask所涉及到的狀態機的時候再說吧。

這個類一部分都是一些初始化的工作，都很輕鬆就能看懂，主要是在

if (useNewApi) {
      System.out.println("新事前："+umbilical.getClass().getName());
      LOG.info("新事前："+umbilical.getClass().getName());
      runNewMapper(job, splitMetaInfo, umbilical, reporter);
    } else {
      System.out.println("舊事前："+umbilical.getClass().getName());
      LOG.info("舊事前："+umbilical.getClass().getName());
      runOldMapper(job, splitMetaInfo, umbilical, reporter);
    }
    LOG.info("事後："+umbilical.getClass().getName());
    System.out.println("事後："+umbilical.getClass().getName());
    //sendSend(umbilical);
    done(umbilical, reporter);

這一部分，判斷使用者是否指定使用新的map方式還是舊的map方式（新的方式和舊的方式主要功能並沒有變化，主要是新的runNewMapper在架構上要比runOldMapper好一點。。。官方說的），這裡我們以runOldMapper為例

InputSplit inputSplit = getSplitDetails(new Path(splitIndex.getSplitLocation()),
            splitIndex.getStartOffset());

這部分程式碼是指，map任務得到job分配給他的需要他來計算的那一部分分片。

RecordReader<INKEY, INVALUE> in = isSkipping() ?
            new SkippingRecordReader<INKEY, INVALUE>(umbilical, reporter, job) :
            new TrackedRecordReader<INKEY, INVALUE>(reporter, job);

這一部分是例項化一個RecordReader，主要用於按照行或者某種方式來讀取自己需要計算的那一個分片。具體取決於這裡被例項化的是哪一個實現類。
然後是hadoop會呼叫使用者編寫的map方法用來處理資料。

再下面是MapOutputCollector這個類，主要用於處理map的輸出。

這裡面有很多方法比較重要，比如sortAndSpill()這個方法。
這個方法主要是被SpillThread呼叫的，
SpillThread這個執行緒在MapOutputBuffer初始化時被啟動，它會一直監視著環形緩衝區當環形緩衝區超過一定值（我記得原始碼裡面好像是寫了是在大部分情況下是百分之八十被佔滿後）的時候將會呼叫sortAndSpill方法，講環形緩衝區中已經計算完畢的資料寫到本地檔案中。

當整個runOldMapper（或者是runNewMapper）完畢後將會執行done()函式傳送事件，通過狀態機變遷完成一些事情（這部分下次會提到）

大概主要的流程也就是這幾個，如果有需要的或者不明白的地方，歡迎發郵件給我[email protected]。