TaskTracker获取并执行map或reduce任务的过程（一）

　　我们知道TaskTracker在默认情况下，每个3秒就行JobTracker发送一个心跳包，也就是在这个心跳包中包含对任务的请求。JobTracker返回给TaskTracker的心跳包中包含有各种action（任务），如果有满足在此TaskTracker上执行的任务的话，该任务也就包含在心跳包的响应中。在TaskTracker端有线程专门等待map或reduce任务，并从队列中取出执行。

1. TaskTracker发送心跳包

　　TaskTracker是作为一个单独的JVM运行的，它启动以后一直处于offerService（）函数中，每隔3秒就执行一次transmitHeartBeat函数，如下所示：

HeartbeatResponse heartbeatResponse = transmitHeartBeat(now);

　　该函数具体代码为：

  HeartbeatResponse transmitHeartBeat(long now) throws IOException {
　　......
    if (status == null) {

      synchronized (this) {

        status = new TaskTrackerStatus(taskTrackerName, localHostname,

                                       httpPort,

                                       cloneAndResetRunningTaskStatuses(

                                         sendCounters),

                                       failures,

                                       maxMapSlots,

                                       maxReduceSlots);

      }

    } //

    // 检查是否可以接受新的任务

    //

    boolean askForNewTask;

    long localMinSpaceStart;

    synchronized (this) {

      askForNewTask =

        ((status.countOccupiedMapSlots() < maxMapSlots ||

          status.countOccupiedReduceSlots() < maxReduceSlots) &&

         acceptNewTasks);

      localMinSpaceStart = minSpaceStart;

    }
......

    HeartbeatResponse heartbeatResponse = jobClient.heartbeat(status,

                                                              justStarted,

                                                              justInited,

                                                              askForNewTask,

                                                              heartbeatResponseId);

......

    return heartbeatResponse;

  }

　　我们从中可以看出，TaskTracker首先创建一个TaskTrackerStatus对象，其中包含有TaskTracker的各种信息，比如，map slot的数目，reducer slot槽的数目，TaskTracker所在的主机名等信息。然后，对TaskTracker的空闲的slot以及磁盘空间进行检查，如果满足相应的条件时，最终就会通过JobClient(为JobTracker的代理)将心跳信息发送给JobTracker，并得到JobTracker的响应HeartbeatResponse。如下所示，JobClient是InterTrackerProtocol的一个实例，而JobTracker实现了InterTrackerProtocol这个接口。

    this.jobClient = (InterTrackerProtocol)

    UserGroupInformation.getLoginUser().doAs(

        new PrivilegedExceptionAction<Object>() {

      public Object run() throws IOException {

        return RPC.waitForProxy(InterTrackerProtocol.class,

            InterTrackerProtocol.versionID,

            jobTrackAddr, fConf);

      }

    });

　　　　那么，TaskTracker怎样通过JobTracker的代理与JobTracker进行通信呢?它是通过RPC调用JobTracker的heartbeat(......)方法而实现的。

2. TaskTracker端获取任务

　　TaskTracker接收到任务后，会将它们放入到相应的LinkedList中，LinkedList实现了List和Queue接口，它是基于链表实现的FIFO的队列。

heartbeatInterval = heartbeatResponse.getHeartbeatInterval();if (actions != null){

          for(TaskTrackerAction action: actions) {

            if (action instanceof LaunchTaskAction) {

              addToTaskQueue((LaunchTaskAction)action);

         ......

          }

        }
　　......

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

　　TaskTracker启动的时候，创建了两个线程：mapLauncher和reduceLauncher，它们分别处理map任务和reduce任务，map任务有mapLauncher负责将其放入到LinkedList中，reduce任务有reducerLauncher负责将其放入到它维护的LinkedList中。

  public void addToTaskQueue(LaunchTaskAction action) {

      synchronized (tasksToLaunch) {

        TaskInProgress tip = registerTask(action, this);

        tasksToLaunch.add(tip);

        tasksToLaunch.notifyAll();

      }

    }

　　mapLauncher或者是reducerLauncher根据接收到的action，创建对应的TaskTracker.TaskInProgress对象，并放入到队列中，唤醒等待的线程进行处理。如下所示，该线程负责从taskToLaunch中获取task，当有空间的slot时，执行这个task。

  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");

          }

.....

          //得到空闲的slot后，启动这个task

          startNewTask(tip);

　　这样，TaskTracker就得到了待处理的任务，具体如何执行请参考下一篇博客。

巴特西

TaskTracker获取并执行map或reduce任务的过程（一）

1. TaskTracker发送心跳包

2. TaskTracker端获取任务

最新文章

热门文章