JobTracker启动流程源码级分析

　　org.apache.hadoop.mapred.JobTracker类是个独立的进程，有自己的main函数。JobTracker是在网络环境中提交及运行MR任务的核心位置。

　　main方法主要代码有两句：

 　　　　 //创建jobTracker对象

         JobTracker tracker = startTracker(new JobConf());

         //启动各个服务，包括JT内部一些重要的服务或者线程

         tracker.offerService();

　　一、startTracker(new JobConf())根据配置文件启动JobTracker，这个方法会调用startTracker(conf, generateNewIdentifier())方法进行启动操作，generateNewIdentifier()将会返回一个以节点当前时间格式化成“yyyyMMddHHmm”的字符串。

startTracker函数是一个静态函数，它调用JobTracker的构造函数生成一个JobTracker类的实例，构造函数主要工作是对一些重要的变量进行初始化，名为result。然后，进行了一系列初始化活动，包括启动RPC server，启动内置的jetty服务器，检查是否需要重启JobTracker等。　
初始化的重要对象包括：
　　1、secretManager：DelegationTokenSecretManager的实例，MR安全管理相关类；
　　2、aclsManager：ACLsManager的实例，作业级别和队列级别的管理和访问权限控制；
　　3、taskScheduler：TaskScheduler的实例，调度器对象，hadoop默认的调度器是FIFO策略的JobQueueTaskScheduler；
　　4、interTrackerServer：Server的实例，RPC Server；
　　5、infoServer：HttpServer的实例，将Job、Task、TaskTracker相关信息显示到WEB前端，封装的是jetty；
　　6、recoveryManager：RecoveryManager的实例，作业恢复管理，即JobTracker启动时，恢复上次停止时正在运行的作业，并恢复各个任务的运行状态；recoveryManager.checkAndAddJob(status)会检查出那些作业需要恢复并放入Set<JobID> jobsToRecover; // set of jobs to be recovered，为后面的recoveryManager.recover()做准备；
　　7、jobHistoryServer：JobHistoryServer的实例，用于查看作业历史信息的Server；
　　8、dnsToSwitchMapping：DNSToSwitchMapping的实例，用于构建集群的网络拓扑结构，它能将节点地址（IP或者host）映射成网络位置。

　　二、 tracker.offerService()

 /**

    * 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();

     //用于发现和清理死掉的TaskTracker

     this.expireTrackersThread = new Thread(this.expireTrackers,

                                           "expireTrackers");

     this.expireTrackersThread.start();

     //用于清理长时间驻留在内存中的已经运行完成的作业信息

     this.retireJobsThread = new Thread(this.retireJobs, "retireJobs");

     this.retireJobsThread.start();

     //用于发现已经被分配给某个TaskTracker但一直未汇报信息的任务

     expireLaunchingTaskThread.start();

     if (completedJobStatusStore.isActive()) {

       completedJobsStoreThread = new Thread(completedJobStatusStore,

                                             "completedjobsStore-housekeeper");

       //将已经运行完成的作业运行信息保存到HDFS上，并提供了一套存取这些信息的API。

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

   }

　　1、首先是不论重启是什么状态都必须要做的recoveryManager.updateRestartCount()更新JobTracker集群重启次数，更新文件${hadoop.tmp.dir}/mapred/system/jobtracker.info。该方法首先判断如果有restartFile(就是前面说的更新文件)，就删除tmpRestartFile(不管存在与否)；如果不存在restartFile而存在tmpRestartFile，则将tmpRestartFile重命名为restartFile；如果两个文件都没有可能是第一次启动也可能是文件丢失了，这时就不用恢复操作了shouldRecover = false，并且创建一个restartFile写入0；再读出restartFile文件的数字，并+1，创建一个tmpRestartFile将增加后的重启次数计数器restartCount写入这个文件，删除restartFile文件，将tmpRestartFile改名为restartFile。这个重启次数存在的目的官方说法是“The whole purpose of this api is to obtain restart counts across restarts to avoid attempt-id clashes.”

　　2、taskScheduler.start()是启动调度器。默认的调度器是JobQueueTaskScheduler，其start()方法如下：

 //JobQueueTaskScheduler类的start方法主要注册了两个非常重要的监听 器：

   //jobQueueJobInProgressListener和eagerTaskInitializationListener。

   //前者是 JobQueueJobInProgressListener类的一个实例，该类以先进先出的方式维持一个JobInProgress的队列，

   //并且监听各 个JobInProgress实例在生命周期中的变化；后者是EagerTaskInitializationListener类的一个实例，

   //该类不断监 听jobInitQueue，一旦发现有新的job被提交（即有新的JobInProgress实例被加入），

   //则立即调用该实例的initTasks方 法，对job进行初始化。

   @Override

   public synchronized void start() throws IOException {

     //调用TaskScheduler.start()方法，实际上没有做任何事情

     super.start();

   //注册一个JobInProgressListerner监听器

     taskTrackerManager.addJobInProgressListener(jobQueueJobInProgressListener);

     eagerTaskInitializationListener.setTaskTrackerManager(taskTrackerManager);

     eagerTaskInitializationListener.start();

     taskTrackerManager.addJobInProgressListener(

         eagerTaskInitializationListener);

   }

　　taskTrackerManager其实就是JobTracker。eagerTaskInitializationListener.start()会启动一个线程始终监控List<JobInProgress> jobInitQueue一旦发现里面有新的JobInProgress就构造一个InitJob线程放入threadPool线程池中运行，该线程通过JobTracker.initJob(JobInProgress job)对Job进行初始化。然后向JobTracker注册eagerTaskInitializationListener。　

　　3、recoveryManager.recover()。JobTracker节点由于意外情况而宕机的话，那么可能有一部分Job正在执行，也有一部分Job被用户成功提交了可还没有开始被调度执行，那么当我们重启JobTracker节点的时候就需要恢复或者重做这些还没有完成的Job。这里要说的是RecoveryManager启动对未完成Job的恢复是在JobTracker节点的主线程中完成的，而且是在JobTracker节点的所有后台线程启动之前，这个调用必须要在所有的未完成的Job被完成之后才返回。也就是说，JobTracker的作业恢复管理器在恢复作业的处理过程中，JobTracker节点不会接受客户端的任何请求，也不接受TaskTracker的任何请求。这个比较复杂以后再讲解。

　　4、refreshHosts()方法会先重新加载mapred.hosts和mapred.hosts.exclude指定的文件中主机信息到相应的Set中；然后从taskTrackers中找出没在mapred.hosts中但在mapred.hosts.exclude中的taskTracker从相关的数据结构中删除此taskTracker。节点均以mapred.hosts和mapred.hosts.exclude中的为准。

　　5、启动一个ExpireTrackers线程会监控trackerExpiryQueue一旦里面TaskTracker有超过10分钟没有心跳的，JobTracker就认为它死了，将其从相关的数据结构trackerToMarkedTasksMap、trackerToJobsToCleanup、trackerToTasksToCleanup以及trackerToTaskMap删除。这在lostTaskTracker(TaskTracker taskTracker)方法中进行。

　　6、启动一个RetireJobs线程，会将jobs中的完成的job存储一定时长后，从taskidToTrackerMap、trackerToTaskMap、taskidToTIPMap、jobs、userToJobsMap(每个用户完成的job数要>100)数据结构中删除。

　　7、启动一个ExpireLaunchingTasks线程，如果一个TaskAttemptID超过10分钟没有回报信息，则JobTracker认为这个task已经失败，从launchingTasks删除相关信息，并将此task状态标注为FAILED。

　　8、启动一个CompletedJobStatusStore线程，默认"mapred.job.tracker.persist.jobstatus.active"是false表示不启动这个线程，如果启用则需要指定保存时间"mapred.job.tracker.persist.jobstatus.hours"(默认是0，不保存)和保存路径"mapred.job.tracker.persist.jobstatus.dir"(默认是/jobtracker/jobsInfo)。如果不启用该线程则所有的作业运行信息全部在内存中，且随着时间及运行任务的增多早期的作业信息会被删除。

　　这样JobTracker就启动了。。。。就等着Client提交Job。。。

　　参考：1、董西成，《Hadoop技术内幕：深入解析MapReduce架构设计与实现原理》

　　　　 2、http://blog.csdn.net/xhh198781/article/details/7354257

巴特西

JobTracker启动流程源码级分析

最新文章

热门文章