Job的任务执行流程之JobCleanup阶段

超、凢脫俗 2022-06-13 09:18 205阅读 0赞

JobTracker节点给TaskTracker节点分配作业任务时是有优先级顺序的，JobTracker节点总是优先分配一个作业的辅助任务，然后在分配作业的正式任务。其中，作业的辅助任务包括：JobSetup、JobCleanup、TaskCleanup三种类型的任务，正式任务包括：Map、Reduce两种任务。当JobTracker节点为一个TaskTracker节点分配了辅助任务的时候就不会为它在分配正式任务了，而给TaskTracker分配的辅助任务中只能有一个任务。这里就有一个问题了，只分配一个辅助任务，那么到底优先分配一个作业的哪种类型的任务呢？JobTracker节点对辅助任务的分配是以作业为单位的，而不是以任务类型为单位的(请好好考虑此处的表述)，这种优先级的顺序如下：

1.Map类型的JobCleanup任务
    2.Map类型的TaskCleanup任务
    3.Map类型的JobSetup任务
    4.Reduce类型的JobCleanup任务
    5.Reduce类型的TaskCleanup任务
    6.Reduce类型的JobSetup任务

JobSetup类型的任务在前面已经详细的阐述过了，所以在本文，笔者将主要讲述作业的JobCleanup任务。当一个作业完成时(成功、失败、kill)，JobTracker节点就会调度这个任务主要来结束作业。

分配一个作业的JobCleanup任务的过程很简单，它的源代码如下：

public Task obtainJobCleanupTask(TaskTrackerStatus tts, int clusterSize, int numUniqueHosts, boolean isMapSlot) throws IOException {
        if(!tasksInited.get()) {
          return null;
        }
        
        //判断此时能够运执行JobCleanup任务
        synchronized(this) {
          if (!canLaunchJobCleanupTask()) {
            return null;
          }
          
          String taskTracker = tts.getTrackerName();
          // Update the last-known clusterSize
          this.clusterSize = clusterSize;
    
          //判断此时能否在该TaskTracker节点上运行JobCleanup任务
          if (!shouldRunOnTaskTracker(taskTracker)) {
            return null;
          }
          
          List<TaskInProgress> cleanupTaskList = new ArrayList<TaskInProgress>();
          if (isMapSlot) {
            cleanupTaskList.add(cleanup[0]);
          } else {
            cleanupTaskList.add(cleanup[1]);
          }
          TaskInProgress tip = findTaskFromList(cleanupTaskList,tts, numUniqueHosts, false);
          if (tip == null) {
            return null;
          }
          
          // Now launch the cleanupTask
          Task result = tip.getTaskToRun(tts.getTrackerName());
          if (result != null) {
            addRunningTaskToTIP(tip, result.getTaskID(), tts, true);
          }
          
          return result;
        }

在调度一个作业的JobCleanup任务的过程中，首先需要判断此时是否能够运行该任务，也即是它需要满足一定的条件，只要满足下列三个条件中的任何一个均可，这三个条件是：

1.作业失败了；

2.作业被kill掉了；

3.同时满足下列四个子条件：

a).作业的状态为：RUNNING或者PREP；

b).该JobCleanup任务没有正在运行同时JobSetup任务完成了；

c).成功完成的Map任务数量与彻底失败的Map任务数量之和等于作业总的Map任务数；

d).成功完成的Reduce任务数量与彻底失败的Reduce任务数量之和等于作业总的Reduce任务数；

从上面的源代码可以看出，一个作业有两个JobCleanup类型的任务，一个是Map型的，一个是Reduce型的，不过，他们完成的功能是一摸一样的，所以是需要调度这两个任务中的一个即可。另外，它同JobSetup、Reduce任务一样，在任何时刻，同时只能有一个实例在TaskTracker节点上执行。

TaskTracker节点接收到JobCleanup类型的任务之后，对其本地化及调度方法同JobSetup任务等是一样的，因为他们都属于Map/Reduce任务，所以都需要JVM实例来完成。JVM实例对任务的执行很简单，这个作业级别的cleanup操作依赖于该作业的输出提交器Outputcommitter，即主要调用它的cleanupJob()方法。对于hadoop内部默认的输出提交其实现FileOutputCommitter，它的cleanupJob()方法主要负责删除该作业在执行过程中创建的一些临时文件及路径。