hadoop源码TaskAttemptID TaskTrackerAction JobTracker，FileOutputCommitter相关

红太狼 2022-03-16 16:10 96阅读 0赞

**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的输出；