Hadoop JobTracker 分析

转自: http://ronxin999.blog.163.com/blog/static/42217920201312152220887/

jobtracker的主要逻辑在offerService方法里面。

主要做了一下几件事：

1 更新jobTracker重启的次数。它简单的创建一个jobtracker信息文件

在jobtracker的系统目录并写道其重启的计数。第一开始,jobtracker写道“0”。在随后的

重启jobtracker取代了与当前计数,计数(老数+ 1)。这个目的是获得重新启动的计数,

以避免尝试id的冲突。如果这个操作失败了。则会等待一定的时间重试，到成功才结束该操作。

2 任务调度器开始工作。该过程需要单独写篇博文来分析其中的原理。待续

3 开始恢复job，通过RecoveryManager的recover()方法实现。该过程需要单独写篇博文来分析其中的原理。

4 更新taskTracker主机列表，把那些不在主机列表中的节点remove掉。主要配置文件为mapred.hosts和

mapred.hosts.exclude

5 开启ExpireTrackers线程，作用是对失去链接TaskTracker线程监控，即长时间未报告的TaskTracker将不会再分配新的Task。因为可能cashed。

6 开启RetireJobs线程，作用是清除那些已经完成很长时间还存在队列里的jobs

7开启expireLaunchingTaskThread线程，作用是停止那些未在超时时间内报告进度的Tasks。

8 开启completedJobsStoreThread线程，作用是把JobInProgress中的job信息存储到DFS中；

提供一些读取状态 信息的方法；是一个守护进程，用于删除DFS中的保存时间超过规定时间的job status

删除。<wbr style="line-height:32px"></wbr>

9 启动interTrackerServer，是一个链接桥梁。端口配置在Configuration中的“mapred.job.tracker”参数，缺省是绑定8012端口。它有两个用途，一是用于接收和处理 TaskTracker的heartbeat等请求，即必须实现InterTrackerProtocol接口及协议。二是用于接收和处理 JobClient的请求，如submitJob，killJob等，即必须实现JobSubmissionProtocol接口及协议。

10 标记状态为running。

上面这10点除了2其他都可以说是JobTracker的周边工作，或者是辅助工作为调度Job。除了上面这10点外，JobTracker的核心工作是接受taskTracker的心跳请求，并为其分配task。以其接收和处理JobClient的请求，如submitJob，killJob等。这是核心工作，需要独立的博文来分析。敬请期待。

下面把这段源码贴下， 可以对着看：

/**
   * 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);
    }
    // refresh the node list as the recovery manager might have added 
    // disallowed trackers
    refreshHosts();
    
    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;
    }
    LOG.info("Starting RUNNING");  
    this.interTrackerServer.join();
    LOG.info("Stopped interTrackerServer");
  }