hadoop源码TaskAttemptID TaskTrackerAction JobTracker，FileOutputCommitter...

gtdfe

hadoop源码TaskAttemptID TaskTrackerAction JobTracker，FileOutputCommitter相关
1,TaskAttemptID代表task attempt，一个task attempt就是一个map/reduce task 的一个实例taskid，而每个TaskAttemptID由两部分组成：TaskID+task序列号
eg:

attempt_200707121733_0003_m_000005_0

代表2007年07月12日17点33分启动的第0003号作业（job）的第00005号map任务的第0号task attempt
2,TaskTrackerAction的类型有，LaunchTaskAction()，KillTaskAction()，KillJobAction()，ReinitTrackerAction()，CommitTaskAction();
/**
   * Ennumeration of various 'actions' that the {@link JobTracker}
   * directs the {@link TaskTracker} to perform periodically.
   *
   */
  public static enum ActionType {
    /** Launch a new task. */
    LAUNCH_TASK,
   
    /** Kill a task. */
    KILL_TASK,
   
    /** Kill any tasks of this job and cleanup. */
    KILL_JOB,
   
    /** Reinitialize the tasktracker. */
    REINIT_TRACKER,

    /** Ask a task to save its output. */
    COMMIT_TASK
  };
  
  /**
   * A factory-method to create objects of given {@link ActionType}.
   * @param actionType the {@link ActionType} of object to create.
   * @return an object of {@link ActionType}.
   */
  public static TaskTrackerAction createAction(ActionType actionType) {
    TaskTrackerAction action = null;
   
    switch (actionType) {
    case LAUNCH_TASK:
      {
        action = new LaunchTaskAction();
      }
      break;
    case KILL_TASK:
      {
        action = new KillTaskAction();
      }
      break;
    case KILL_JOB:
      {
        action = new KillJobAction();
      }
      break;
    case REINIT_TRACKER:
      {
        action = new ReinitTrackerAction();
      }
      break;
    case COMMIT_TASK:
      {
        action = new CommitTaskAction();
      }
      break;
    }

    return action;
  }
3,JobTracker中相关：
           offerService()
                   |--taskScheduler.start();
                   |--recoveryManager.recover();
通过RecoveryManager的recover()方法恢复Job

                   |-- this.expireTrackersThread.start();
检测已经失效的TaskTracker节点。

// Used to expire TaskTrackers that have gone down

                   |-- this.retireJobsThread.start();
清除那些已经完成很长时间仍然存在队列中的job

                   |--expireLaunchingTaskThread.start();
停止那些在超时时间内未报告进度的Tasks。

A thread to timeout tasks that have been assigned to task trackers,
   * but that haven't reported back yet.

                   |--completedJobsStoreThread.start();
                   |--this.interTrackerServer.start();/ start the inter-tracker server once the jt is ready
  

  /**
   * Run forever
   */
  public void offerService() throws InterruptedException, IOException {
    // Prepare for recovery. This is done irrespective of the status of restart
    // flag.
    while (true) {
      try {
        recoveryManager.updateRestartCount();
        break;
      } catch (IOException ioe) {
        LOG.warn("Failed to initialize recovery manager. ", ioe);
        // wait for some time
        Thread.sleep(FS_ACCESS_RETRY_PERIOD);
        LOG.warn("Retrying...");
      }
    }

    taskScheduler.start();
   
    //  Start the recovery after starting the scheduler
    try {
      recoveryManager.recover();
    } catch (Throwable t) {
      LOG.warn("Recovery manager crashed! Ignoring.", t);
    }
   
    this.expireTrackersThread = new Thread(this.expireTrackers,
                                          "expireTrackers");
    this.expireTrackersThread.start();
    this.retireJobsThread = new Thread(this.retireJobs, "retireJobs");
    this.retireJobsThread.start();
    expireLaunchingTaskThread.start();

    if (completedJobStatusStore.isActive()) {
      completedJobsStoreThread = new Thread(completedJobStatusStore,
                                            "completedjobsStore-housekeeper");
      completedJobsStoreThread.start();
    }

    // start the inter-tracker server once the jt is ready
    this.interTrackerServer.start();
   
    synchronized (this) {
      state = State.RUNNING;
    }
4, Job的五种状态
public static final int RUNNING = 1;
  public static final int SUCCEEDED = 2;
  public static final int FAILED = 3;
  public static final int PREP = 4;

  public static final int KILLED = 5;

5，jobStatus信息:

  this.jobid = jobid;
     this.setupProgress = setupProgress;
     this.mapProgress = mapProgress;
     this.reduceProgress = reduceProgress;
     this.cleanupProgress = cleanupProgress;
     this.runState = runState;

6，FileOutputCommitter相关

setupJob：Hadoop初始化时设置job的输出；

commitJob：当job完成时，清除job的输出，这个方法在反馈回来的job状态为SUCCEEDED时调用；

cleanupJob：job结束后清除job的输出；

// do the clean up of temporary directory

abortJob：当job的返回状态是FAILED或KILLED时，执行该函数，用于终止作业的输出；

setupTask：设置task的输出；

// FileOutputCommitter's setupTask doesn't do anything. Because the
       // temporary task directory is created on demand when the
       // task is writing.

needsTaskCommit：检测task是否需要提交；

commitTask：将task的输出移到作业的输出目录；

// Move the task outputs to their final place

// Delete the temporary task-specific output directory

abortTask：取消task的输出；