JobTracker之作业恢复与权限管理机制（源码分析第四篇）

红太狼 2022-04-10 01:17 182阅读 0赞

**（一）概述**

JobTracker是整个MapReduce计算框架的核心组件，它在操作系统中独占一个主服务进程。一个MapReduce应用程序在hadoop中被抽象为一个作业，并且为了实现更好的分布式计算将作业分割为若干个task(分治算法思想)。其中在整个MapReduce框架中JobTracker在充当“管理者”角色，负责作业的控制和系统资源的管理工作，说的具体点就是负责作业的分解，各个作业、其分解之后的子任务和taskTracker（通过心跳机制）的状态的监控和更新。之所有要对作业和任务进行监控，第一：是为了随时查看作业和任务的进度以及完成情况，第二：JobTracker对作业和任务的监控并且保存了其在运行时的信息，这些信息可以为任务调度提供决策依据。第三：作业和任务在运行过程中往往不是“一帆风顺”的，需要一定的容错机制以保证作业和任务高效地正常完成（MapReduce的容错机制还是比较完善的），而JobTracker对作业、任务以及taskTracker实行实时监控可以及时的发现出现故障的taskTracker、作业以及任务，从而及时的启动相应的容错措施。本文主要是以hadoop-1.0.0源码为依据分析JobTracker的启动过程，以及一些重要初始化对象,作业的恢复与权限的管理。

**（二）具体分析**

**1、启动入口**

JobTracker作为单独的后台进程，其由JobTracker类中的main方法启动。

/**
       * Start the JobTracker process.  This is used only for debugging.  As a rule,
       * JobTracker should be run as part of the DFS Namenode process.
       */
      public static void main(String argv[]
                              ) throws IOException, InterruptedException {
        StringUtils.startupShutdownMessage(JobTracker.class, argv, LOG);
                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                         
        try {
          if(argv.length == 0) {
            JobTracker tracker = startTracker(new JobConf());//创建JobTracker对象
            tracker.offerService();//启动各个服务线程
          }
          else {
            if ("-dumpConfiguration".equals(argv[0]) && argv.length == 1) {
              dumpConfiguration(new PrintWriter(System.out));
            }
            else {
              System.out.println("usage: JobTracker [-dumpConfiguration]");
              System.exit(-1);
            }
          }
        } catch (Throwable e) {
          LOG.fatal(StringUtils.stringifyException(e));
          System.exit(-1);
        }
      }

JobTracker通过offerService方法启动JobTracker里面负责各种功能给您的线程，其中包括作业恢复线程、资源清理线程、任务调度器线程等，后面将详细分析。我们进入到startTracker(newJobConf())方法中，

public static JobTracker startTracker(JobConf conf, String identifier)
      throws IOException, InterruptedException {
        DefaultMetricsSystem.initialize("JobTracker");
        JobTracker result = null;
        while (true) {
          try {
            result = new JobTracker(conf, identifier);
            result.taskScheduler.setTaskTrackerManager(result);
            break;
          } catch (VersionMismatch e) {
    ......

首先我们知道JobTracker会通过任务调度器（TaskScheduler）来对任务进行调度，任务调度器就是在这个时候被指定的。

**2、JobTracker重要对象，作业恢复和权限管理。**

接下来我们看看JobTracker里面的一些重要对象的作业，先看看源码中都有哪些重要对象(只列出部分)：

private final TaskScheduler taskScheduler;
 ......
 Map<String,Integer>uniqueHostsMap = new ConcurrentHashMap<String, Integer>();
   ExpireTrackers expireTrackers = new ExpireTrackers();
   Thread expireTrackersThread = null;
   RetireJobs retireJobs = new RetireJobs();
   Thread retireJobsThread = null;
   final int retiredJobsCacheSize;
   ExpireLaunchingTasks expireLaunchingTasks = new ExpireLaunchingTasks();
   Thread expireLaunchingTaskThread = new Thread(expireLaunchingTasks,
                                                 "expireLaunchingTasks");
   CompletedJobStatusStore completedJobStatusStore = null;
   Thread completedJobsStoreThread = null;
   RecoveryManager recoveryManager;
   JobHistoryServer jobHistoryServer;
 ......
 final HttpServer infoServer;
   int infoPort;
   Server interTrackerServer;
 ......
 private final ACLsManager aclsManager;
 ....
 private QueueManager queueManager;
 ...

（1）taskScheduler

TaskScheduler接口实现类，默认的作业任务调度器，负责任务调度和资源管理。

（2）aclsManager

ACLsManager类的实例，此对象用于实现作业队列和作业的管理和访问权限。其中作业权限包括查看作业和修改作业，队列权限包括管理队列和向队列中提交作业。该类中有多个

checkAccess()方法的重载，主要作用检测用户是否有提交作业权限、检查是否可以查看或者修改作业信息、查看任务日志信息等。下面看一种类型的重载代码:

void checkAccess(String jobId, UserGroupInformation callerUGI,
         String queue, Operation operation, String jobOwner,
         AccessControlList jobAcl) throws AccessControlException {
      ......
       if (isMRAdmin(callerUGI)) {
         AuditLogger.logSuccess(user, operation.name(), targetResource);
         return;
       }
       if (operation == Operation.SUBMIT_JOB) {//检查是否有提交作业权限
         // This is strictly queue operation(not a job operation)
         if (!queueManager.hasAccess(queue, operation.qACLNeeded, callerUGI)) {
          ......
         } else {
           AuditLogger.logSuccess(user, operation.name(), targetResource);
           return;
         }
       }
       // Check if callerUGI is queueAdmin, jobOwner or part of job-acl.
       // queueManager and queue are null only when called from
       // TaskTracker(i.e. from TaskLogServlet) for the operation VIEW_TASK_LOGS.
       // Caller of this method takes care of checking if callerUGI is a
       // queue administrator for that operation.
       if (operation == Operation.VIEW_TASK_LOGS) {//检查是否有查看任务日志权限
     ......
         }
       } else if (queueManager.hasAccess(queue,operation.qACLNeeded,callerUGI) ||jobACLsManager.checkAccess(callerUGI,
    operation.jobACLNeeded,jobOwner, jobAcl)) {
         AuditLogger.logSuccess(user, operation.name(), targetResource);
         return;
       }
       ......
     }

此外，我们进入到ACLsManager类中，可以看到其对作业队列和作业的权限管理是分别分给JobACLsManager和QueueManager这个2个类去实现，其具体说明：

<table> 
 <tbody> 
 <tr> 
 <td style="border-color:#dddddd;"> 
 <ul style="list-style-type:square;"> 
 <li>JobACLsManager</li> 
 </ul> &nbsp; 作业权限控制类。主要负责设置作业的查看和修改权限。 &nbsp; 其中设置查看权限主要体现在：可以限制别的用户访问该作业以及该作业的任务的Counter、该作业所任务所在TaskTracker在50030上的日志信息以及在50070中生成的job.xml文件。此选项可以在作业提交的时候设置mapreduce.job.acl-view-job参数。 设置修改权限体现在：可以防止其他用户修改自己提交作业的信息，比如：kill掉作业、更改优先级、kill 掉任务等。此选项可以再作业提交时设置mapreduce.job.acl-modify-job参数。 
 <ul style="list-style-type:square;"> 
 <li>QueueManager </li> 
 </ul> &nbsp; 队列权限控制类。主要负责作业的提交权限（指定那些用户可以向队列中添加新的作业）、作业管理权限（指定那些用户可以对该队列中的作业进行管理）。其可以再配置文件mapred-queue-acls.xml中指定，参数分别是mapred.queue.&lt;queue-name&gt;.acl-submit-job和mapred.queue.&lt;queue-name&gt;.acl-administer-jobs。</td> 
 </tr> 
 </tbody> 
</table>

**小结一下权限管理aclsManager的作用（看图）：**

[![wKiom1LUnV7iKlmgAAKQpd70KOw004.jpg][]][wKiom1LUnV7iKlmgAAKQpd70KOw004.jpg]

（3）recoveryManager、jobHistoryServer、JobHistory类

recoveryManager和jobHistoryServer分别是RecoveryManager类和JobHistoryServer类的实例。这两个对象主要职责就是对作业进行恢复工作(注意：JobHistory类负责JobTracker关键事件日志 记录日志，它是作业恢复的基础)，以解决JobTracker存在的单点故障问题。如果集群管理员开启了作业恢复功能（设置参数mapred.jobtracker.restart.recover为true），那么当JobTracker产生故障重启（注意：JobTracker的重启过程中，各个TaskTracker仍然活着的）后会自动先检测是否存在需要恢复运行状态的作业。如果有，则可以通过日志恢复这些作业的运行状态，并重新调度那么些没有运行的tasks,当然也包括了已经产生部分结果的tasks。下面分别介绍这两个类具体的协调工作：

<table> 
 <tbody> 
 <tr> 
 <td style="border-color:#dddddd;"> 
 <ul style="list-style-type:circle;"> 
 <li>jobHistoryServer 用于查询作业历史信息的Server，它里面置有Web容器可以提供在50030界面查看。其实这个类在MapReduce框架中有两种不同的启动方式，第一种方式：将其嵌入到JobTracker中，由JobTracker对其进行启动初始化，但是这种方式是必须要求JobHistoryServer所使用的地址和JobTracker一样；第二种方式：用户可以对其进行单独启动（JobHistoryServer类中有main方法提供单独启动），这时它是作为系统的一个daemon进行，但是其地址可以不和JobTracker一样。代码中的注释是这样说的：</li> 
 </ul><pre><code>*&nbsp;mapred.job.history.server.http.address&nbsp;is&nbsp;address&nbsp;to&nbsp;which&nbsp;history&nbsp;web
*&nbsp;&nbsp;server&nbsp;is&nbsp;bound&nbsp;to.&nbsp;If&nbsp;operating&nbsp;in&nbsp;embedded&nbsp;mode,&nbsp;the&nbsp;hostname&nbsp;on
*&nbsp;&nbsp;history&nbsp;address&nbsp;has&nbsp;to&nbsp;be&nbsp;same&nbsp;as&nbsp;the&nbsp;job&nbsp;tracker&nbsp;host&nbsp;name
*
*&nbsp;mapred.job.history.server.embedded&nbsp;(default&nbsp;is&nbsp;true)&nbsp;will&nbsp;cause&nbsp;job&nbsp;tracker
*&nbsp;&nbsp;to&nbsp;init&nbsp;history&nbsp;server,&nbsp;else&nbsp;the&nbsp;server&nbsp;need&nbsp;to&nbsp;be&nbsp;started&nbsp;as&nbsp;a
*&nbsp;&nbsp;separate&nbsp;daemon&nbsp;process
&nbsp;*****************************************************************/
public&nbsp;class&nbsp;JobHistoryServer&nbsp;{
......</code></pre> 再看看JobHistoryServer作业单独的daemon进程启动的方式代码：<pre><code>/**
&nbsp;&nbsp;&nbsp;*&nbsp;Start&nbsp;job&nbsp;history&nbsp;server&nbsp;as&nbsp;an&nbsp;independent&nbsp;process
&nbsp;&nbsp;&nbsp;*
&nbsp;&nbsp;&nbsp;*&nbsp;@param&nbsp;args&nbsp;-&nbsp;Command&nbsp;line&nbsp;arguments
&nbsp;&nbsp;&nbsp;*/
&nbsp;&nbsp;public&nbsp;static&nbsp;void&nbsp;main(String[]&nbsp;args)&nbsp;{
&nbsp;&nbsp;&nbsp;&nbsp;StringUtils.startupShutdownMessage(JobHistoryServer.class,&nbsp;args,&nbsp;LOG);
&nbsp;&nbsp;&nbsp;&nbsp;try&nbsp;{
&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;JobHistoryServer&nbsp;server&nbsp;=&nbsp;new&nbsp;JobHistoryServer(new&nbsp;JobConf());
&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;server.start();
&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;server.join();
&nbsp;&nbsp;&nbsp;&nbsp;}&nbsp;catch&nbsp;(Throwable&nbsp;e)&nbsp;{
&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;LOG.fatal(StringUtils.stringifyException(e));
&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;System.exit(-1);
&nbsp;&nbsp;&nbsp;&nbsp;}
&nbsp;&nbsp;}</code></pre> 
 <ul style="list-style-type:circle;"> 
 <li>JobHistory类</li> 
 </ul> &nbsp; &nbsp;该类的主要用于MapReduce框架中的一些关键事件的日志记录。对于作业而已体现在：提交作业记录、创建作业记录、作业开始运行记录、作业结束运行记录、kill掉作业记录、作业失败记录等等。对于任务而与主要体现在：创建任记录、开始运行任务记录、任务运行成功记录、任务运行失败记录、任务被kill掉记录等等。这些记录作业的恢复机制的基础，也是整个MapReduce框架中的容错机制的重要组成部分。看看几个JobHistory类中具体的日志记录方法：<pre>//Log&nbsp;task&nbsp;attempt&nbsp;killed&nbsp;event.
public&nbsp;static&nbsp;void&nbsp;logKilled(TaskAttemptID&nbsp;taskAttemptId,
&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;long&nbsp;timestamp,&nbsp;String&nbsp;hostName,
&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;String&nbsp;error,&nbsp;String&nbsp;taskType)&nbsp;{
&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;JobID&nbsp;id&nbsp;=&nbsp;taskAttemptId.getJobID();
&nbsp;&nbsp;&nbsp;&nbsp;......
&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;}
&nbsp;&nbsp;&nbsp;&nbsp;}
//Log&nbsp;task&nbsp;attempt&nbsp;failed&nbsp;event.
&nbsp;&nbsp;public&nbsp;static&nbsp;void&nbsp;logFailed(TaskAttemptID&nbsp;taskAttemptId,
&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;long&nbsp;timestamp,&nbsp;String&nbsp;hostName,
&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;String&nbsp;error,&nbsp;String&nbsp;taskType)&nbsp;{
&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;JobID&nbsp;id&nbsp;=&nbsp;taskAttemptId.getJobID();
......&nbsp;}
//Log&nbsp;finish&nbsp;time&nbsp;of&nbsp;map&nbsp;task&nbsp;attempt.
public&nbsp;static&nbsp;void&nbsp;logFinished(TaskAttemptID&nbsp;taskAttemptId,
&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;long&nbsp;finishTime,
&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;String&nbsp;hostName,
&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;String&nbsp;taskType,
&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;String&nbsp;stateString,
&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;Counters&nbsp;counter)&nbsp;{
&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;JobID&nbsp;id&nbsp;=&nbsp;taskAttemptId.getJobID();
......}</pre> 
 <ul style="list-style-type:circle;"> 
 <li>recoveryManager</li> 
 </ul>用于通过日志信息恢复作业状态,当JobTracker启动时首先会去检查有没有需要恢复的作业（前提是开启作业恢复模式），如果有，就恢复上次停止时正在运行的作业，并且恢复各个任务的运行状态。 小结一下recoveryManager、jobHistoryServer、JobHistory类协调工作的情况： <a href="http://s3.51cto.com/wyfs02/M00/11/8E/wKiom1LU7KOhDVraAAKGy9JjoaY245.jpg" rel="nofollow"><img src="http://s3.51cto.com/wyfs02/M00/11/8E/wKiom1LU7KOhDVraAAKGy9JjoaY245.jpg" title="ww (2).jpg" width="600" height="361" style="width:600px;" alt="wKiom1LU7KOhDVraAAKGy9JjoaY245.jpg"></a></td> 
 </tr> 
 </tbody> 
</table>

（4）infoTrackerServer

org.apache.hadoop.ipc.Server接口实现，作为RPC Server负责JobTracker和各个TaskTracker心跳处理。具体表现为：判断TaskTracker是否活着、接收TaskTracker申请任务的请求、获得各个节点的资源使用情况以及任务运行情况。

**（三）总结**

本文主要分析了JobTracker的启动方式和入口、作业的恢复机制、权限管理机制。其中作业的恢复机制中分别提到了RecoveryManager、JobHistory、JobHistory的作用和他们在对作业恢复时的协调关系。另外，在权限管理机制中提到了AclsManager类以及它的两个重要变量类JobACLsManager和QueueManager的作用，他们分别负责作业和队列权限的管理。

由于JobTracker类的的代码量比较大（有5376行代码），由于篇幅限制，本文只分析了JobTracker中很小的一部分功能，在接下来的一篇blog中继续分析探讨JobTracker里边各种线程和心跳相关机制。

#  #

# \---------------------------------------**hadoop源码分析系列**\-\----------------------------------------------------------------------------------------------------------- #

[hadoop作业分片处理以及任务本地性分析（源码分析第一篇）][hadoop]

[hadoop作业提交过程分析（源码分析第二篇）][hadoop 1]

[hadoop作业初始化过程详解（源码分析第三篇）][hadoop 2]

[JobTracker之作业恢复与权限管理机制（源码分析第四篇）][JobTracker]

[JobTracker之辅助线程和对象映射模型分析（源码分析第五篇）][JobTracker 1]

# \---------------\---------------\---------------\---------------------------------------\--------------------------------------------------------------------------------------------- #

参考文献：

\[1\]《Hadoop技术内幕：深入解析MapReduce架构设计与实现原理》

\[2\] [http://hadoop.apache.org/][http_hadoop.apache.org]

[wKiom1LUnV7iKlmgAAKQpd70KOw004.jpg]: /images/20220404/651f5a66cbdf4e70bcbbc483b4450319.png
[hadoop]: http://zengzhaozheng.blog.51cto.com/8219051/1347777
[hadoop 1]: http://zengzhaozheng.blog.51cto.com/8219051/1349155
[hadoop 2]: http://zengzhaozheng.blog.51cto.com/8219051/1350161
[JobTracker]: http://zengzhaozheng.blog.51cto.com/8219051/1351578
[JobTracker 1]: http://zengzhaozheng.blog.51cto.com/8219051/1352280
[http_hadoop.apache.org]: http://hadoop.apache.org/