MapReduce剖析笔记之六：TaskTracker初始化任务并启动JVM过程

在上面一节我们分析了JobTracker调用JobQueueTaskScheduler进行任务分配，JobQueueTaskScheduler又调用JobInProgress按照一定顺序查找任务的流程，获得了任务之后，将任务封装为TaskTrackerAction数组返回的整个过程。TaskTracker通过心跳响应接收到了这个数组。本节我们继续分析，TaskTracker拿到了这个数组之后，如何对任务进行处理的。

1，TaskTracker在其方法offerService中，将得到的任务加入队列：

        TaskTrackerAction[] actions = heartbeatResponse.getActions();
。。。。。。。
        if (actions != null){ 
          for(TaskTrackerAction action: actions) {
            if (action instanceof LaunchTaskAction) {
              addToTaskQueue((LaunchTaskAction)action);
            } else if (action instanceof CommitTaskAction) {
              CommitTaskAction commitAction = (CommitTaskAction)action;
              if (!commitResponses.contains(commitAction.getTaskID())) {
                commitResponses.add(commitAction.getTaskID());
              }
            } else {
              addActionToCleanup(action);
            }
          }
        }

可见，返回的TaskTrackerAction对象有多种可能，最典型的是LaunchTaskAction，即启动一个任务；另外还有CommitTaskAction等。TaskTrackerAction是一个抽象类，其实现总共有5个，如下所示：

  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
  };

LaunchTaskAction表示启动任务，内部包含一个Task对象，即要启动的任务对象；

CommitTaskAction表示让一个正在执行中的任务提交其输出结果，其内部包含一个TaskAttemptID对象，对于某些Job来说，只有Map阶段而没有Reduce阶段，此时可以让其直接提交结果；

KillJobAction表示杀死属于该Job的任意任务，内部包含一个JobID；

KillTaskAction表示杀死一个任务，内部包含一个TaskAttemptID；

ReinitTrackerAction表示让TaskTracker重新初始化，无需带什么参数，只要随着心跳响应返回，对应的TaskTracker自然知道应该重新初始化。

我们对最典型的LaunchTaskAction进行分析，其内部的Task是一个抽象类，主要有MapTask和ReduceTask两种实现。在上面的代码中，addToTaskQueue将LaunchTaskAction按照Map或Reduce类型分别加入到不同的队列：

  private void addToTaskQueue(LaunchTaskAction action) {
    if (action.getTask().isMapTask()) {
      mapLauncher.addToTaskQueue(action);
    } else {
      reduceLauncher.addToTaskQueue(action);
    }
  }

mapLauncher和reduceLauncher都是TaskLauncher类型的对象，顾名思义用于Task的启动。TaskLauncher继承了Thread类，因此是一个线程类。其加入队列的代码为：

    public void addToTaskQueue(LaunchTaskAction action) {
      synchronized (tasksToLaunch) {
        TaskInProgress tip = registerTask(action, this);
        tasksToLaunch.add(tip);
        tasksToLaunch.notifyAll();
      }
    }

tasksToLaunch是一个任务链表：LinkedList<TaskInProgress>()，可见，我们再一次又看见了TaskInProgress，心跳响应时，双方交互的实际上是Task对象，自己内部维护的是TaskInProgress对象。通过registerTask这个方法，将其转换回来。其实其转换代码也简单：

  private TaskInProgress registerTask(LaunchTaskAction action, 
      TaskLauncher launcher) {
    Task t = action.getTask();
    LOG.info("LaunchTaskAction (registerTask): " + t.getTaskID() +
             " task's state:" + t.getState());
    TaskInProgress tip = new TaskInProgress(t, this.fConf, launcher);
    synchronized (this) {
      tasks.put(t.getTaskID(), tip);
      runningTasks.put(t.getTaskID(), tip);
      boolean isMap = t.isMapTask();
      if (isMap) {
        mapTotal++;
      } else {
        reduceTotal++;
      }
    }
    return tip;
  }

在上面的注册代码中，创建了一个TaskInProgress对象，之后加入到两个映射表中：

  Map<TaskAttemptID, TaskInProgress> tasks = new HashMap<TaskAttemptID, TaskInProgress>();
  /**
   * Map from taskId -> TaskInProgress.
   */
  Map<TaskAttemptID, TaskInProgress> runningTasks = null;

映射表的索引是目前的任务ID，值即为刚刚创建的TaskInProgress对象。

之后，加入到tasksToLaunch对象中，并执行notifyAll方法，唤醒TaskLauncher线程（该线程可能处于睡眠等待中）处理队列里面的任务。

接下来，另一个线程，即TaskLauncher的run方法会处理这个对象。心跳线程即TaskTracker的主线程返回，继续等待执行后续的心跳。

2，TaskLauncher的run方法接着执行：

run方法的前半部分是从队列中获取任务：

          synchronized (tasksToLaunch) {
            while (tasksToLaunch.isEmpty()) {
              tasksToLaunch.wait();
            }
            //get the TIP
            tip = tasksToLaunch.remove(0);
            task = tip.getTask();
            LOG.info("Trying to launch : " + tip.getTask().getTaskID() + 
                     " which needs " + task.getNumSlotsRequired() + " slots");
          }

如果为空，则调用wait进行等待；如果不为空，自然有人会通知自己（即notifyAll），之后，取出第一个任务并删除队列中的第一个元素。wait和notifyAll是JAVA任意一个对象都会具有的方法。

取出任务之后，需要判断是否有资源可用，使用IntWritable numFreeSlots进行判断，如果没有足够资源，则一直会等待：

            while (numFreeSlots.get() < task.getNumSlotsRequired()) {
。。。。。。。。。。。。。
              numFreeSlots.wait();
            }

而numFreeSlots的更新是TaskLauncher另一个方法中进行的：

    public void addFreeSlots(int numSlots) {
      synchronized (numFreeSlots) {
        numFreeSlots.set(numFreeSlots.get() + numSlots);
        assert (numFreeSlots.get() <= maxSlots);
        LOG.info("addFreeSlot : current free slots : " + numFreeSlots.get());
        numFreeSlots.notifyAll();
      }
    }

而该方法又被另一个方法调用：

    private synchronized void releaseSlot() {
      if (slotTaken) {
        if (launcher != null) {
          launcher.addFreeSlots(task.getNumSlotsRequired());
        }
        slotTaken = false;
      } else {
        // wake up the launcher. it may be waiting to block slots for this task.
        if (launcher != null) {
          launcher.notifySlots();
        }
      }
    }

releaseSlot表示释放资源，又是在kill和reportTaskFinished方法中调用：

    void reportTaskFinished(boolean commitPending) {
      if (!commitPending) {
        taskFinished();
        releaseSlot();
      }
      notifyTTAboutTaskCompletion();
    }

也就是说，当任务被杀掉或者执行完毕时，会释放资源，进而发出通告。

如果获得了足够的资源，则该变量会变化：

 numFreeSlots.set(numFreeSlots.get() - task.getNumSlotsRequired());

之后，如果任务可以执行（如果不可以执行还需要回收资源），进入startNewTask方法。在该方法里，会创建一个新线程进行任务的启动工作，上一个线程只是从队列里取出任务，判断资源是否充足等功能。启动了新线程后，它的使命结束了，返回去继续等待队列LinkedList<TaskInProgress>()中是否出现了新的TaskInProgress。

3，startNewTask创建的线程继续执行：

  void startNewTask(final TaskInProgress tip) throws InterruptedException {
    Thread launchThread = new Thread(new Runnable() {
      @Override
      public void run() {
        try {
          RunningJob rjob = localizeJob(tip);
          tip.getTask().setJobFile(rjob.getLocalizedJobConf().toString());
          launchTaskForJob(tip, new JobConf(rjob.getJobConf()), rjob); 
        } catch (Throwable e) {
。。。。。
          try {
            tip.kill(true);
            tip.cleanup(false, true);
          } catch (IOException ie2) {
。。。。。。
        }
      }
    });
    launchThread.start();
  }

该线程中，首先执行localizeJob方法。这个方法用于初始化Job的目录。首先，执行addTaskToJob方法，将任务加入到正在运行作业的对象中，runningJobs是TaskTracker中的一个变量，记录了当前这个TaskTracker上面运行了哪些Job。

  Map<JobID, RunningJob> runningJobs = new TreeMap<JobID, RunningJob>();

runningJobs里面保存了Job ID和RunningJob之间的对应关系，RunningJob是一个类，记录了当前正在初始化的任务的信息。主要包含以下信息：

    private JobID jobid; 
    private JobConf jobConf;
    private Path localizedJobConf;
    // keep this for later use
    volatile Set<TaskInProgress> tasks;
    //the 'localizing' and 'localized' fields have the following
    //state transitions (first entry is for 'localizing')
    //{false,false} -> {true,false} -> {false,true}
    volatile boolean localized;
    boolean localizing;
    boolean keepJobFiles;
    UserGroupInformation ugi;
    FetchStatus f;
    TaskDistributedCacheManager distCacheMgr;

即Job ID、Job Conf即配置参数、Job配置文件路径、该Job包含的任务集合（当前TaskTracker内的）以及一些用户权限等信息。

接下来需要对Job所在目录进行定位，使用initializeJob方法：

上一节我们在分析任务调度的时候，有一个obtainNewMapTaskCommon方法，该方法通过查找任务后，执行addRunningTaskToTIP方法，并调用addRunningTask方法，即创建MapTask或ReduceTask对象。此时会将任务的目录信息传进来，而最终一个任务的目录信息是在TaskInProgress的构造方法中传进来的。我们再回顾下Job的初始化过程，有以下代码：

    maps = new TaskInProgress[numMapTasks];
    for(int i=0; i < numMapTasks; ++i) {
      inputLength += splits[i].getInputDataLength();
      maps[i] = new TaskInProgress(jobId, jobFile, 
                                   splits[i], 
                                   jobtracker, conf, this, i, numSlotsPerMap);
    }

这里的jobFile即为任务所在的目录，实际上任务所在目录来源于作业所在目录，因此一个JobInProgress对象在创建的时候就会将目录信息记录下来，实际上，最终的文件目录是在Job初始化时的submitJobInternal方法传进来的。我们回顾以下关键代码：

        Path jobStagingArea = JobSubmissionFiles.getStagingDir(JobClient.this,
            jobCopy);
        JobID jobId = jobSubmitClient.getNewJobId();
        Path submitJobDir = new Path(jobStagingArea, jobId.toString());

这个目录即Staging目录，是根据一定规则自动生成的，最终位于HDFS的某个路径中。

我们回到initializeJob这个方法。因为Job提交时其实已经确定了目录，所以以下代码就是找到这个目录对应的文件系统：

    final JobID jobId = t.getJobID();
    final Path jobFile = new Path(t.getJobFile());
    final Configuration conf = getJobConf();
    FileSystem userFs = getFS(jobFile, jobId, conf);

这个文件系统就是HDFS中作业所在的Staging目录。Staging是集结地、驻留区的意思。用户在提交作业的时候，会将所需的jar文件、配置文件（conf、xml）、其它依赖的lib包、要处理数据的Split信息文件等全部上传到这个地方。

获得了这个目录后，接下来，TaskTracker的任务初始化过程会将上述处于HDFS中的文件拷贝到本地（如Linux）文件系统。使用localizeJobConfFile方法：

    FileStatus status = null;
    long jobFileSize = -1;
    try {
      status = userFs.getFileStatus(jobFile);
      jobFileSize = status.getLen();
    } catch(FileNotFoundException fe) {
      jobFileSize = -1;
    }
    Path localJobFile = lDirAlloc.getLocalPathForWrite(getPrivateDirJobConfFile(user,jobId.toString()), jobFileSize, fConf);

    // Download job.xml
    userFs.copyToLocalFile(jobFile, localJobFile);

lDirAlloc是一个LocalDirAllocator对象，主要用于本地目录的操作。getLocalPathForWrite方法：

  public Path getLocalPathForWrite(String pathStr, long size, 
                                   Configuration conf,
                                   boolean checkWrite) throws IOException {
    AllocatorPerContext context = obtainContext(contextCfgItemName);
    return context.getLocalPathForWrite(pathStr, size, conf, checkWrite);
  }

AllocatorPerContext对象通过 obtainContext方法获得。这里面涉及到一个参数contextCfgItemName，该参数是LocalDirAllocator对象在创建的时候传入的：该对象本身是TaskTracker的一个变量：

    this.localDirAllocator = new LocalDirAllocator("mapred.local.dir");

  public LocalDirAllocator(String contextCfgItemName) {
    this.contextCfgItemName = contextCfgItemName;
  }

之后调用 context.getLocalPathForWrite方法，首先执行confChanged方法。在该方法里，有以下代码：

      String newLocalDirs = conf.get(contextCfgItemName);
      if (!newLocalDirs.equals(savedLocalDirs)) {
        String[] localDirs = conf.getStrings(contextCfgItemName);
        localFS = FileSystem.getLocal(conf);

此处，通过调用文件系统相关方法，会创建"mapred.local.dir"所指定的本地目录。

创建好了本地目录后，执行文件拷贝操作copyToLocalFile：

  public void copyToLocalFile(boolean delSrc, Path src, Path dst)
    throws IOException {
    FileUtil.copy(this, src, getLocal(getConf()), dst, delSrc, getConf());
  }

即job.xml文件被从HDFS中拷贝到TaskTracker所在的本地文件系统。该目录由配置文件"mapred.local.dir"指定。

另外，在这之前还有一些拷贝关于用户操作权限等文件：localizeJobTokenFile（与localizeJobConfFile过程类似）。

接下来对拷贝到本地的配置文件进行解析，得到JobConf对象：

final JobConf localJobConf = new JobConf(localJobFile);

并将这个信息加入到缓存中，因为localJobConf是根据某个任务拷贝过来的job.xml解析的，如果后面需要再用，则可以使用该缓存信息。TaskTracker里面维护了一个TrackerDistributedCacheManager distributedCacheManager对象，该对象里面维护了一个任务缓存：

  private Map<JobID, TaskDistributedCacheManager> jobArchives =
    Collections.synchronizedMap(
        new HashMap<JobID, TaskDistributedCacheManager>());

以及一个TaskController taskController。

首先将上面解析的任务配置参数加入到缓存中：

  public TaskDistributedCacheManager newTaskDistributedCacheManager(JobID jobId,
                                 Configuration taskConf) throws IOException {
    TaskDistributedCacheManager result = new TaskDistributedCacheManager(this, taskConf);
    jobArchives.put(jobId, result);
    return result;
  }

TrackerDistributedCacheManager以及TaskDistributedCacheManager 有什么意义呢？每个TaskTracker创建时会创建一个TrackerDistributedCacheManager对象，其主要作用是管理该机器上所有任务的Cache文件。而当任务在初始化时，会由TrackerDistributedCacheManager产生一个TaskDistributedCacheManager对象（如上面代码示例），用于管理本任务的Cache文件。有什么好处呢？比如，一个TaskTracker上可能会有一个Job的多个Task，这样，某个TaskDistributedCacheManager已经获取处理了某些Cache文件，则其他任务无需再处理。这里所谓的Cache文件，指的是从HDFS文件系统拷贝到本地文件系统的那些文件，本地的Linux系统相对于HDFS来说可以看做一种缓存，因为不能每次要访问某个参数，或者一个机器上的所有任务要访问Job信息都去HDFS中去获取和解析吧，这就是Distributed Cache分布式缓存的含义，因为有很多机器会拷贝这个文件。对于某个TaskTracker而言，通过拷贝到本地，不仅加速了后面的处理时间，也为其他任务做出了贡献。只是需要注意的是，如何维护本地的缓存信息和HDFS中的一致是需要解决的问题。Hadoop里设计了一个CacheFile类，里面包含了时间戳等信息，只要比较一下该时间戳以及HDFS中文件时间戳是否一致即可。

获得了缓存对象后，本地的Job配置会有一些更新，更新后写回到本地的配置文件：

          // Set some config values
          localJobConf.set(JobConf.MAPRED_LOCAL_DIR_PROPERTY,
              getJobConf().get(JobConf.MAPRED_LOCAL_DIR_PROPERTY));
          if (conf.get("slave.host.name") != null) {
            localJobConf.set("slave.host.name", conf.get("slave.host.name"));
          }
          resetNumTasksPerJvm(localJobConf);
          localJobConf.setUser(t.getUser());

          // write back the config (this config will have the updates that the
          // distributed cache manager makes as well)
          JobLocalizer.writeLocalJobFile(localJobFile, localJobConf);

接下来进入到taskController的initializeJob方法，进行后续的初始化。TaskController是一个抽象类，其实现是DefaultTaskController和LinuxTaskController。DefaultTaskController是默认的任务控制器，LinuxTaskController在Linux系统上使用，在用户权限设置等方面有些不同，可能是利用Linux的用户管理等等进行任务管理，以后有时间再研究，现在先来看默认的DefaultTaskController。

其initializeJob方法声明如下，可以看出，该方法从TaskTracker的私有空间拷贝证书文件到Job的私有空间；创建Job工作目录，下载job.jar文件并解压得到要执行的任务的.class文件，并更新配置。之后，创建一个JobConf配置对象，设置分布式缓存和用户日志目录等

  /**
   * This routine initializes the local file system for running a job.
   * Details:
   * <ul>
   * <li>Copies the credentials file from the TaskTracker's private space to
   * the job's private space </li>
   * <li>Creates the job work directory and set 
   * {@link TaskTracker#JOB_LOCAL_DIR} in the configuration</li>
   * <li>Downloads the job.jar, unjars it, and updates the configuration to 
   * reflect the localized path of the job.jar</li>
   * <li>Creates a base JobConf in the job's private space</li>
   * <li>Sets up the distributed cache</li>
   * <li>Sets up the user logs directory for the job</li>
   * </ul>
   * This method must be invoked in the access control context of the job owner 
   * user. This is because the distributed cache is also setup here and the 
   * access to the hdfs files requires authentication tokens in case where 
   * security is enabled.
   * @param user the user in question (the job owner)
   * @param jobid the ID of the job in question
   * @param credentials the path to the credentials file that the TaskTracker
   * downloaded
   * @param jobConf the path to the job configuration file that the TaskTracker
   * downloaded
   * @param taskTracker the connection to the task tracker
   * @throws IOException
   * @throws InterruptedException
   */
  @Override
  public void initializeJob(String user, String jobid, 
                            Path credentials, Path jobConf, 
                            TaskUmbilicalProtocol taskTracker,
                            InetSocketAddress ttAddr
                            ) 

其输入参数主要有：用户、JobID、证书目录、Job配置文件目录、一个TaskUmbilicalProtocol对象，以及TaskTracker的IP地址信息，这里的TaskUmbilicalProtocol对象需要细说一下，TaskUmbilicalProtocol是和InterTrackerProtocol等等类似的一个RPC接口，我们前面分析过，再来总结一下：

1）JobClient和JobTracker之间存在一个RPC接口JobSubmissionProtocol，JobClient调用这个接口，JobTracker实现了这个接口。

2）JobTracker和TaskTracker之间存在一个RPC接口InterTrackerProtocol，TaskTracker调用这个接口，JobTracker实现了这个接口。

TaskUmbilicalProtocol也类似，不过是TaskTracker实现的，那么，谁来掉用呢？

Umbilical大致是脐带的意思，直观含义是连接胎儿和母体之间的重要通道。TaskTracker要启动一个Map或Reduce任务的时候，是通过启动另外一个JAVA虚拟机去运行这个任务，用户提交作业时，就是JobClient和JobTracker在打交道，JobTracker对这个作业初始化的过程会创建HDFS目录，将jar文件、分析处理的文件（即FileInputFormat.addInputPath里面指定的），生成Split信息文件等统统放到这个目录。在某个TaskTracker与自己心跳时，从作业队列挑选一些合适的任务，让TaskTracker运行，两者之间交互了Task的信息，TaskTracker接收到这个信息后，去HDFS目录里面将Job文件拷贝到本地，同时jar文件也拷贝下来，进行解压以后怎么运行呢？TaskTracker会启动一个JAVA虚拟机，这个JAVA虚拟机当然也是一个进程，不过称为TaskTracker进程的子进程，这个子进程和TaskTracker进程之间也需要交互一些信息，比如汇报状态等等，此时就导致了TaskUmbilicalProtocol这个RPC接口的出现。我们看看他有哪些方法：

  JvmTask getTask(JvmContext context) throws IOException;

  boolean statusUpdate(TaskAttemptID taskId, TaskStatus taskStatus,
      JvmContext jvmContext) throws IOException, InterruptedException;
  
  void reportDiagnosticInfo(TaskAttemptID taskid, String trace,
      JvmContext jvmContext) throws IOException;
 
  void reportNextRecordRange(TaskAttemptID taskid, SortedRanges.Range range,
      JvmContext jvmContext) throws IOException;

  boolean ping(TaskAttemptID taskid, JvmContext jvmContext) throws IOException;

  void done(TaskAttemptID taskid, JvmContext jvmContext) throws IOException;
  
  void commitPending(TaskAttemptID taskId, TaskStatus taskStatus,
      JvmContext jvmContext) throws IOException, InterruptedException;  

  boolean canCommit(TaskAttemptID taskid, JvmContext jvmContext) throws IOException;

  void shuffleError(TaskAttemptID taskId, String message, JvmContext jvmContext) throws IOException;
  
  void fsError(TaskAttemptID taskId, String message, JvmContext jvmContext) throws IOException;

  void fatalError(TaskAttemptID taskId, String message, JvmContext jvmContext)  throws IOException;

  MapTaskCompletionEventsUpdate getMapCompletionEvents(JobID jobId, 
                                                       int fromIndex, 
                                                       int maxLocs,
                                                       TaskAttemptID id,
                                                       JvmContext jvmContext)  throws IOException;

  void updatePrivateDistributedCacheSizes(org.apache.hadoop.mapreduce.JobID jobId,long[] sizes) throws IOException;

可见，主要是状态更新、错误信息上报、周期地ping、任务完成通知等等。

所以，现在可以理解，上面initializeJob方法中传入的TaskUmbilicalProtocol对象其实就是TaskTracker.this。

我们接下里分析initializeJob方法。

这个方法的主要功能上面已经进行了简单描述，来看一下细节。首先需要做的就是将TaskTracker从HDFS拷贝过来的文件进行二次拷贝，拷贝到作业所在的目录。这里使用了JobLocalizer类，这个类会创建以下目录：

/**
 * Internal class responsible for initializing the job, not intended for users.
 * Creates the following hierarchy:
 *   <li>$mapred.local.dir/taskTracker/$user</li>
 *   <li>$mapred.local.dir/taskTracker/$user/jobcache</li>
 *   <li>$mapred.local.dir/taskTracker/$user/jobcache/$jobid/work</li>
 *   <li>$mapred.local.dir/taskTracker/$user/jobcache/$jobid/jars</li>
 *   <li>$mapred.local.dir/taskTracker/$user/jobcache/$jobid/jars/job.jar</li>
 *   <li>$mapred.local.dir/taskTracker/$user/jobcache/$jobid/job.xml</li>
 *   <li>$mapred.local.dir/taskTracker/$user/jobcache/$jobid/jobToken</li>
 *   <li>$mapred.local.dir/taskTracker/$user/distcache</li>
 */

主要的代码是：

    FileSystem localFs = FileSystem.getLocal(getConf());
    JobLocalizer localizer = new JobLocalizer((JobConf)getConf(), user, jobid);
    localizer.createLocalDirs();
    localizer.createUserDirs();
    localizer.createJobDirs();
    JobConf jConf = new JobConf(jobConf);
    localizer.createWorkDir(jConf);
    //copy the credential file
    Path localJobTokenFile = lDirAlloc.getLocalPathForWrite(
        TaskTracker.getLocalJobTokenFile(user, jobid), getConf());
    FileUtil.copy(localFs, credentials, localFs, localJobTokenFile, false, getConf());
    //setup the user logs dir
    localizer.initializeJobLogDir();
    // Download the job.jar for this job from the system FS
    // setup the distributed cache
    // write job acls
    // write localized config
    localizer.localizeJobFiles(JobID.forName(jobid), jConf, localJobTokenFile, taskTracker);

这里比较重要的最后一行：localizeJobFiles，jar文件就是在这里处理的，进去看看。

执行的第一行代码就是localizeJobJarFile，前面已经把job.xml等文件拷贝到本地了，jar文件还没拷贝，此处就是做这个工作的。

主要代码为：

      // Here we check for five times the size of jarFileSize to accommodate for
      // unjarring the jar file in the jars directory
      Path localJarFile = lDirAlloc.getLocalPathForWrite(JARDST, 5 * jarFileSize, ttConf);
      //Download job.jar
      userFs.copyToLocalFile(jarFilePath, localJarFile);
      localJobConf.setJar(localJarFile.toString());
      // Also un-jar the job.jar files. We un-jar it so that classes inside
      // sub-directories, for e.g., lib/, classes/ are available on class-path
      RunJar.unJar(new File(localJarFile.toString()),
          new File(localJarFile.getParent().toString()));
      FileUtil.chmod(localJarFile.getParent().toString(), "ugo+rx", true);

在获得了本地的jar目录后，从HDFS中进行拷贝，并利用RunJar的unJar方法进行解压。JAVA里有JarFile这个类用于JAR打包和解压。顺便说一下，jar文件采用的就是ZIP格式，JarFile其实就是利用ZIP的解压方法，而JAVA中本身也有ZIP的解压方法，ZIP利用的是LZ系列字典压缩算法以及Huffman、游程编码等算法的融合。JarFile继承于java.util.zip.ZipFile这个类。

至此，TaskController的initializeJob方法结束，返回TaskTracker的initializeJob方法，并继续返回localizeJob方法。RunningJob这个对象初始化完毕，改变下面的值：

 rjob.localized = true;

并向其他线程执行通告：

      synchronized (rjob) {
        if (rjob.localizing) {
          rjob.localizing = false;
          rjob.notifyAll();
        }
      }

    synchronized (runningJobs) {
      runningJobs.notify(); //notify the fetcher thread
    }

localizeJob方法执行完毕，之后返回startNewTask方法。接下来执行launchTaskForJob方法：

  protected void launchTaskForJob(TaskInProgress tip, JobConf jobConf,
                                RunningJob rjob) throws IOException {
    synchronized (tip) {
      jobConf.set(JobConf.MAPRED_LOCAL_DIR_PROPERTY,
                  localStorage.getDirsString());
      tip.setJobConf(jobConf);
      tip.setUGI(rjob.ugi);
      tip.launchTask(rjob);
    }
  }

首先是设置一些任务参数，之后调用launchTask方法，在launchTask方法里，因为前面涉及的都是Job相关的配置信息，到了此时，Task本身的配置信息需要进行设置：

  /**
   * Localize the given JobConf to be specific for this task.
   */
  public void localizeConfiguration(JobConf conf) throws IOException {
    conf.set("mapred.tip.id", taskId.getTaskID().toString()); 
    conf.set("mapred.task.id", taskId.toString());
    conf.setBoolean("mapred.task.is.map", isMapTask());
    conf.setInt("mapred.task.partition", partition);
    conf.set("mapred.job.id", taskId.getJobID().toString());
  }

比如Task ID、是否是Map任务等等。

设置Task的状态为运行状态：

        if (this.taskStatus.getRunState() == TaskStatus.State.UNASSIGNED) {
          this.taskStatus.setRunState(TaskStatus.State.RUNNING);
        }

接下来会创建一个TaskRunner对象，TaskRunner 对象本身是TaskInProgress中的一个变量，其创建则是在MapTask或ReduceTask中实现的。

  @Override
  public TaskRunner createRunner(TaskTracker tracker, 
                                 TaskTracker.TaskInProgress tip,
                                 TaskTracker.RunningJob rjob
                                 ) throws IOException {
    return new MapTaskRunner(tip, tracker, this.conf, rjob);
  }

TaskRunner是一个抽象类，主要有MapTaskRunner和ReduceTaskRunner两种实现。该抽象类继承于Thread，是一个线程类，并且在TaskRunner中实现了run方法。MapTaskRunner在创建过程中，利用父类构造方法，将一系列参数传进去：

  public TaskRunner(TaskTracker.TaskInProgress tip, TaskTracker tracker, 
                    JobConf conf, TaskTracker.RunningJob rjob
                    ) throws IOException {
    this.tip = tip;
    this.t = tip.getTask();
    this.tracker = tracker;
    this.conf = conf;
    this.mapOutputFile = new MapOutputFile();
    this.mapOutputFile.setConf(conf);
    this.jvmManager = tracker.getJvmManagerInstance();
    this.localdirs = conf.getLocalDirs();
    taskDistributedCacheManager = rjob.distCacheMgr;
  }

比如Map任务之后的目录等，这里还有一个jvmManager对象，该对象是TaskTracker中的一个对象，每个TaskTracker一个，在TaskTracker初始化的时候会创建出来：

  public JvmManager(TaskTracker tracker) {
    mapJvmManager = new JvmManagerForType(tracker.getMaxCurrentMapTasks(), true, tracker);
    reduceJvmManager = new JvmManagerForType(tracker.getMaxCurrentReduceTasks(), false, tracker);
  }

其中：JvmManagerForType的构造方法如下，第一个参数是要管理的JAVA虚拟机的个数，而tracker.getMaxCurrentMapTasks()获得的就是为本TaskTracker配置的最大Map槽数。

    public JvmManagerForType(int maxJvms, boolean isMap, 
        TaskTracker tracker) {
      this.maxJvms = maxJvms;
      this.isMap = isMap;
      this.tracker = tracker;
      sleeptimeBeforeSigkill =
        tracker.getJobConf().getLong(DELAY_BEFORE_KILL_KEY,
                                     DEFAULT_SLEEPTIME_BEFORE_SIGKILL);
    }

创建好了TaskRunner之后，即启动新线程：

        setTaskRunner(task.createRunner(TaskTracker.this, this, rjob));
        this.runner.start();

launchTask这个方法启动了TaskRunner之后，则返回launchTaskForJob方法，继续返回startNewTask方法。对tasksToLaunch中的其他TaskInProgress方法重复执行上面的初始化操作。startNewTask创建的局部线程已经结束。如果有新的任务到达，TaskLauncher会启动新的线程来执行上面的工作，下面进入到TaskRunner的run方法执行。

4，TaskRunner的run方法接力执行：

这个方法为JVM的启动提供命令行生成、环境变量、类库路径等信息。首先创建Task工作目录，注意上面的JobLocalizer类创建了一系列Job目录，这里创建的是Task相关的目录。

      final File workDir =
      new File(new Path(localdirs[rand.nextInt(localdirs.length)], 
          TaskTracker.getTaskWorkDir(t.getUser(), taskid.getJobID().toString(), 
          taskid.toString(),
          t.isTaskCleanupTask())).toString());
      
      String user = tip.getUGI().getUserName();

之后调用getClassPaths方法获取解压后的.class文件路径

  private static List<String> getClassPaths(JobConf conf, File workDir,
      TaskDistributedCacheManager taskDistributedCacheManager)
      throws IOException {
    // Accumulates class paths for child.
    List<String> classPaths = new ArrayList<String>();
    boolean userClassesTakesPrecedence =  conf.getBoolean(MAPREDUCE_USER_CLASSPATH_FIRST,false);
    if (!userClassesTakesPrecedence) {
      // start with same classpath as parent process
      appendSystemClasspaths(classPaths);
    }
    // include the user specified classpath
    appendJobJarClasspaths(conf.getJar(), classPaths);
    // Distributed cache paths
    classPaths.addAll(taskDistributedCacheManager.getClassPaths());
    // Include the working dir too
    classPaths.add(workDir.toString());
    if (userClassesTakesPrecedence) {
      // parent process's classpath is added last
      appendSystemClasspaths(classPaths);
    }
    return classPaths;
  }

总之，一个JAVA程序运行需要很多库目录，比如本地目录、JRE目录等，首先加入系统目录：

  private static void appendSystemClasspaths(List<String> classPaths) {
    for (String c : System.getProperty("java.class.path").split(
        SYSTEM_PATH_SEPARATOR)) {
      classPaths.add(c);
    }
  }

appendJobJarClasspaths方法加入到Job的jar文件所在的本地目录（已经从HDFS中拷贝下来了）。

之后，创建JAVA虚拟机运行所需要的参数：

      //  Build exec child JVM args.
      Vector<String> vargs = getVMArgs(taskid, workDir, classPaths, logSize);

说白了，就是产生一堆“java xxx -D ...”等等这个命令，比如一些参数示例，虚拟机内存占多大等等：

    //  <property>
    //    <name>mapred.map.child.java.opts</name>
    //    <value>-Xmx 512M -verbose:gc -Xloggc:/tmp/@taskid@.gc 
    //           -Dcom.sun.management.jmxremote.authenticate=false 
    //           -Dcom.sun.management.jmxremote.ssl=false 
    //    </value>
    //  </property>

首先找到JAVA虚拟机那个可执行文件在哪个目录（安装目录/bin），然后加进去，这自然是第一步。

    File jvm =                                  // use same jvm as parent
      new File(new File(System.getProperty("java.home"), "bin"), "java");
    vargs.add(jvm.toString());

另外，就是一些库文件目录等等，比如：

    Path childTmpDir = createChildTmpDir(workDir, conf, false);
    vargs.add("-Djava.io.tmpdir=" + childTmpDir);

    // Add classpath.
    vargs.add("-classpath");
    String classPath = StringUtils.join(SYSTEM_PATH_SEPARATOR, classPaths);
    vargs.add(classPath);

关键的是把要执行的class的类名写上：

vargs.add(Child.class.getName());  // main of Child

可以看出，执行的是Child这个类，这个类有一个Main方法。我们后面再看。

获得了JAVA运行参数后，继续获取环境变量：

      Map<String, String> env = new HashMap<String, String>();
      errorInfo = getVMEnvironment(errorInfo, user, workDir, conf, env, taskid,
                                   logSize);

 比如LD_LIBRARY_PATH等等这些环境变量。

之后，执行launchJvmAndWait方法：

  void launchJvmAndWait(List <String> setup, Vector<String> vargs, File stdout,
      File stderr, long logSize, File workDir)
      throws InterruptedException, IOException {
    jvmManager.launchJvm(this, jvmManager.constructJvmEnv(setup, vargs, stdout,
        stderr, logSize, workDir, conf));
    synchronized (lock) {
      while (!done) {
        lock.wait();
      }
    }
  }

里面的核心方法是jvmManager.launchJvm：

  public void launchJvm(TaskRunner t, JvmEnv env
                        ) throws IOException, InterruptedException {
    if (t.getTask().isMapTask()) {
      mapJvmManager.reapJvm(t, env);
    } else {
      reduceJvmManager.reapJvm(t, env);
    }
  }

mapJvmManager和reduceJvmManager都是一个JvmManagerForType对象，前面提过，在TaskTracker初始化的时候会进行创建。执行JvmManagerForType的reapJvm方法，该方法中，有一个变量spawnNewJvm记录了是否应该启动一个新的JVM。spawn是产卵的意思，该标志的含义很明显，如果不满足一些条件，那么是不能启动虚拟机的，那么有哪些条件呢？

JvmManagerForType对象里面有几个映射表：

    //Mapping from the JVM IDs to running Tasks
    Map <JVMId,TaskRunner> jvmToRunningTask = new HashMap<JVMId, TaskRunner>();
    //Mapping from the tasks to JVM IDs
    Map <TaskRunner,JVMId> runningTaskToJvm = new HashMap<TaskRunner, JVMId>();
    //Mapping from the JVM IDs to Reduce JVM processes
    Map <JVMId, JvmRunner> jvmIdToRunner =  new HashMap<JVMId, JvmRunner>();
    //Mapping from the JVM IDs to process IDs
    Map <JVMId, String> jvmIdToPid = new HashMap<JVMId, String>();

jvmToRunningTask记录了JVMID和TaskRunner对象的关系；runningTaskToJvm则反过来记录；jvmIdToRunner记录了JVMID和JvmRunner的关系。注意这里有几个不同的对象：TaskRunner代表了要运行的任务；JvmRunner代表了要启动的虚拟机。

在判断spawnNewJvm变量时，首先获得当前已经运行的JVM：

int numJvmsSpawned = jvmIdToRunner.size();

如果numJvmsSpawned < maxJvms，那么表明目前启动的JVM还没达到最大值，可以启动：spawnNewJvm=true；

如果numJvmsSpawned >= maxJvms，则表明目前的数量已经至少等于最大可允许的JVM数量（=配置的Map或Reduce Slot数目），这种情况下按理来说可能不能启动了，不过也不一定。这里会对每一个JVM进行逐一判断，找找看，看哪个JVM适合处理这个任务。这是什么逻辑呢？有两种可能，一种可能是利用已经启动的JVM，判断代码如下：

          //look for a free JVM for this job; if one exists then just break
          if (jId.equals(jobId) && !jvmRunner.isBusy() && !jvmRunner.ranAll()){
            setRunningTaskForJvm(jvmRunner.jvmId, t); //reserve the JVM
            LOG.info("No new JVM spawned for jobId/taskid: " + 
                     jobId+"/"+t.getTask().getTaskID() +
                     ". Attempting to reuse: " + jvmRunner.jvmId);
            return;
          }

如果一个JVM内的任务与要启动任务的Job ID一样（jId.equals(jobId)），且已经不忙了，即处于空闲状态（!jvmRunner.isBusy()），并且所有任务没有执行完毕（!jvmRunner.ranAll()），此时则将该任务（TaskRunner代表了要运行的任务）直接让该虚拟机执行，执行setRunningTaskForJvm方法：

    synchronized public void setRunningTaskForJvm(JVMId jvmId, TaskRunner t) {
      jvmToRunningTask.put(jvmId, t);
      runningTaskToJvm.put(t,jvmId);
      jvmIdToRunner.get(jvmId).setBusy(true);
    }

除了上面这种情况，还有一种情况就是看看能不能杀掉一个JVM，判断代码如下：

          if ((jId.equals(jobId) && jvmRunner.ranAll()) ||
              (!jId.equals(jobId) && !jvmRunner.isBusy())) {
            runnerToKill = jvmRunner;
            spawnNewJvm = true;
          }

即两者Job ID相同且JVM已经执行完毕了，此时可以杀掉这个JVM来执行新任务；或者两者Job ID不同且那个JVM处于空闲状态。这说明，一个新的任务并不是一定要启动一个新的JVM，可能会利用已有的JVM继续运行。

如果spawnNewJvm=true，则表示可以启动新的JVM：

      if (spawnNewJvm) {
        if (runnerToKill != null) {
          LOG.info("Killing JVM: " + runnerToKill.jvmId);
          killJvmRunner(runnerToKill);
        }
        spawnNewJvm(jobId, env, t);
        return;
      }

接下来我们看看spawnNewJvm这个方法，其代码较简单：

    private void spawnNewJvm(JobID jobId, JvmEnv env,  
        TaskRunner t) {
      JvmRunner jvmRunner = new JvmRunner(env, jobId, t.getTask());
      jvmIdToRunner.put(jvmRunner.jvmId, jvmRunner);
      jvmRunner.setDaemon(true);
      jvmRunner.setName("JVM Runner " + jvmRunner.jvmId + " spawned.");
      setRunningTaskForJvm(jvmRunner.jvmId, t);
      LOG.info(jvmRunner.getName());
      jvmRunner.start();
    }

首先创建一个JvmRunner对象，表示要运行的JVM，将该JVM记录下来与JVMID与JVMRunner的关系。JVMID是一个JobID加上随机生成的整数组合成的一个对象。因为JvmRunner对象是一个线程类，所以最后一句启动了一个新线程：JVMRunner。此时，TaskRunner线程类的工作结束。

5，JVMRunner的run方法继续执行，其run方法也很简单，就一句话：

      @Override
      public void run() {
        try {
          runChild(env);
        } catch (InterruptedException ie) {
          return;
        } catch (IOException e) {
。。。。。。。。。。。
        } catch (Throwable e) {
。。。。。。。。。
        } finally {
          jvmFinished();
        }
      }

在方法runChild里，关键的一句是:

                exitCode = tracker.getTaskController().launchTask(user,
                    jvmId.jobId.toString(), taskAttemptIdStr, env.setup,
                    env.vargs, env.workDir, env.stdout.toString(),
                    env.stderr.toString());

于是进入DefaultTaskController的launchTask方法，该方法的作用是创建所有Task所需的目录，启动子JVM。

里面关键的代码是：

      // get the JVM command line.
      String cmdLine = 
        TaskLog.buildCommandLine(setup, jvmArguments,
            new File(stdout), new File(stderr), logSize, true);

      // write the command to a file in the
      // task specific cache directory
      // TODO copy to user dir
      Path p = new Path(allocator.getLocalPathForWrite(
          TaskTracker.getPrivateDirTaskScriptLocation(user, jobId, attemptId),
          getConf()), COMMAND_FILE);

      String commandFile = writeCommand(cmdLine, rawFs, p);
      rawFs.setPermission(p, TaskController.TASK_LAUNCH_SCRIPT_PERMISSION);

cmdLin即生成要启动JVM的命令行字符串；第二句话是创建一个文件COMMAND_FILE，其值是COMMAND_FILE = "taskjvm.sh"，即创建一个脚本文件，之后将命令行字符串写入该脚本文件，并设置权限。

最后，做好了这些准备工作后，执行脚本：

      shExec = new ShellCommandExecutor(new String[]{
          "bash", commandFile},
          currentWorkDirectory);
      shExec.execute();

ShellCommandExecutor是一个脚本命令的执行类，execute方法将执行以下代码：

    /** Execute the shell command. */
    public void execute() throws IOException {
      this.run();    
    }

  /** check to see if a command needs to be executed and execute if needed */
  protected void run() throws IOException {
    if (lastTime + interval > System.currentTimeMillis())
      return;
    exitCode = 0; // reset for next run
    runCommand();
  }

因此，进入runCommand方法，该方法使用了JAVA中提供的一个类：java.lang.ProcessBuilder，这个类用于创建操作系统进程。

首先设置环境变量等：

    if (environment != null) {
      builder.environment().putAll(this.environment);
    }
    if (dir != null) {
      builder.directory(this.dir);
    }

然后到了启动JVM的最关键一步，启动进程：

    process = builder.start();

等待进程运行：

      // wait for the process to finish and check the exit code
      exitCode  = process.waitFor();

接下来就是新进程的运行，启动的是Child这个类的Main方法：

  public static void main(String[] args) throws Throwable {
    LOG.debug("Child starting");
。。。。。。

这个执行较长，我们留作下一节分析。

总结一下，本节主要分析了JobTracker将任务调度给TaskTracker后，主要执行了5步操作，对应于5个线程：

1）TaskTracker调用addToTaskQueue加入TaskLauncher（mapLauncher和reduceLauncher）的队列。在这个过程中，会创建对应的TaskInProgress对象加入到队列List<TaskInProgress> tasksToLaunch中，然后返回（这一部分操作是由TaskTracker的心跳线程执行的）；

2）TaskLauncher这个线程类的run会不停地等待新的TaskInProgress，如果出现新的任务，进入startNewTask方法，startNewTask会创建一个新线程，TaskLauncher这个线程类返回去等待新的TaskInProgress任务；

3）startNewTask中的局部线程：从HDFS拷贝相应文件至本地等操作，该线程还会创建一个TaskRunner线程类，接着由其run方法执行；

4）TaskRunner线程类的run方法，负责获取JVM所需的参数信息，又启动JvmRunner这个线程类的run方法；

5）JvmRunner线程类的run方法，负责将JVM启动所需的命令写入一个脚本文件，执行该脚本文件，在另外一个进程中执行了Child的Main方法。

至于该Main进行了什么操作，留作后续分析。