Hadoop1.2.1源码解析系列：JobTracker（一）——JobTracker初始化

JobTracker是hadoop的mapreduce框架中最重要的一个类，这个类负责整个集群的作业控制和资源管理。

JobTracker的启动是在用户启动hadoop集群时启动的，也就是在hadoop启动完之后使用jps命令看到的jobTracker进程，启动代码是在start-mapred.sh中。启动JobTracker是通过调用JobTracker的main()方法启动。下面一步一步慢慢分析JobTracker整个类的代码，先从启动开始。

1.JobTracker.main()：JobTracker tracker = startTracker(new JobConf())启动JobTracker。

2.JobTracker.startTracker()：最终调用的是startTracker(JobConf conf, String identifier, boolean initialize)这个方法，这里注意下initialize传过来的是false。调用result = new JobTracker(conf, identifier)获得一个JobTracker对象。在实例化时最终调用的是JobTracker(final JobConf conf, String identifier, Clock clock, QueueManager qm)
这个构造函数，qm是简单的new QueueManager(new Configuration(conf))得到的。下面看一下QueueManager。

3.QueueManager：

public QueueManager(Configuration conf) {
    checkDeprecation(conf);
    conf.addResource(QUEUE_ACLS_FILE_NAME);
    
    // Get configured ACLs and state for each queue
    aclsEnabled = conf.getBoolean("mapred.acls.enabled", false);

    queues.putAll(parseQueues(conf)); 
  }

checkDeprecation(conf)这个方法主要检查在mapred-site.xml、hdfs-site.xml、core-site.xml中是否出现有关queue的配置，有则发出警告。接着conf.addResource(QUEUE_ACLS_FILE_NAME)为configuration添加mapred-queue-acls.xml配置文件，这个才是正确配置有关queue的参数的地方，所有有关queue的配置都应放入该文件。aclsEnabled = conf.getBoolean("mapred.acls.enabled", false)读取配置文件判断是否启用acl功能，即访问控制功能，默认是false。parseQueues(conf)从配置文件解析每个queue的信息，queue的权限包括作业提交权限和作业管理权限，分别由mapred.queue.queueName.acl-submit-job参数和mapred.queue.queueName.acl-administer-jobs参数决定，参数中的值是用户名或者用户组名。parseQueues(conf)就是解析每个queue哪有用户拥有作业提交权限和哪些用户拥有作业管理权限，具体是根据逗号分隔mapred.queue.queueName.acl-submit-job参数和mapred.queue.queueName.acl-administer-jobs参数值。最后所有的队列权限信息都被添加到queues对象中。

4.JobTracker()：回到JobTracker构造函数。InetSocketAddress addr = getAddress(conf)这句是获得配置文件中的mapred.job.tracker值，并根据该值新建一个InetSocketAddress对象。SecurityUtil.login(conf, JT_KEYTAB_FILE, JT_USER_NAME, localMachine)这句有点像是使用ssh登陆jobTracker所在主机，这里应该就是在搭建hadoop集群时需要设置ssh无密码访问的原因。secretManager = new DelegationTokenSecretManage()是新建一个MapReduce安全管理相关的类。secretManager.startThreads()以后台线程的方法启动secretManager。接下来是一堆参数，略过。其中有些参数后面会用过，TASKTRACKER_EXPIRY_INTERVAL（作业终止间隔），RETIRE_JOB_INTERVAL（作业完成清除间隔），RETIRE_JOB_CHECK_INTERVAL（作业完成清除任务检查间隔），retiredJobsCacheSize（保留的已完成作业数量），MAX_COMPLETE_USER_JOBS_IN_MEMORY（是否将完成的任务保留在hdfs中），这些参数会在后面说到。initializeTaskMemoryRelatedConfig()方法是初始化一些有关内存的参数值。this.hostsReader = new HostsFileReader(conf.get("mapred.hosts", ""),conf.get("mapred.hosts.exclude", ""))(Read the hosts/exclude files to restrict access to the jobtracker)，涉及到两个参数：mapred.hosts和mapred.hosts.exclude，看以参考mapred-default.xml关于这两个参数的描述，大致意思是这两个参数决定哪些node能够访问jobTracker以及哪些node不能访问jobTracker，null则无任何限制，这是一种安全策略，可以防止非法机器访问jobTracker。aclsManager = new ACLsManager(conf, new JobACLsManager(conf), queueManager)，新建一个作业级别和队列级别的管理和访问权限控制对象，所以将queueManager对象传递给aclsManager，同时new一个JobACLsManager对象，queueManager负责队列级别的管理和访问权限控制，而JobACLsManager对象负责作业级别的管理和访问权限控制。

5.JobTracker()：继续分析JobTracker构造函数。clusterMap = (NetworkTopology) ReflectionUtils.newInstance(conf.getClass("net.topology.impl", NetworkTopology.class,
NetworkTopology.class), conf)，根据注释"Create network topology"可以看出是用来构建网络拓扑结构的。接下来是创建一个taskScheduler对象，Class<? extends TaskScheduler> schedulerClass = conf.getClass("mapred.jobtracker.taskScheduler",JobQueueTaskScheduler.class, TaskScheduler.class); taskScheduler = (TaskScheduler) ReflectionUtils.newInstance(schedulerClass, conf)，具体使用的任务调度方案则是由mapred.jobtracker.taskScheduler设置，hadoop默认的作业调度器是JobQueueTaskScheduler，也就是遵循FIFO原则的作业调度器，同时hadoop还自带了FairScheduler和CapacityScheduler两个调度器，在我的另一篇文章中已经介绍了调度器的启动（针对FairScheduler的），这里不进行叙述了。

if (conf.getBoolean(
          ServiceAuthorizationManager.SERVICE_AUTHORIZATION_CONFIG, false)) {
      PolicyProvider policyProvider = 
          (PolicyProvider)(ReflectionUtils.newInstance(
              conf.getClass(PolicyProvider.POLICY_PROVIDER_CONFIG, 
                  MapReducePolicyProvider.class, PolicyProvider.class), 
              conf));
        ServiceAuthorizationManager.refresh(conf, policyProvider);
    }

根据其上的注释感觉有点像是服务级别的安全策略，默认是关闭的。

int handlerCount = conf.getInt("mapred.job.tracker.handler.count", 10);
    this.interTrackerServer = 
      RPC.getServer(this, addr.getHostName(), addr.getPort(), handlerCount, 
          false, conf, secretManager)

这个mapred.job.tracker.handler.count决定jobTracker启动多少个handler用来接收rpc请求，默认是10，这里是一个hadoop的优化点。下面看看RPC.getServer()方法。

6.RPC.getServer()：调用new Server(instance, conf, bindAddress, port, numHandlers, verbose, secretManager)获得一个Server对象，在新建Server对象时会新建一个listener = new Listener()和一个responder = new Responder()对象，用于监听rpc请求和用于发送rpc请求结果。

public Server(Object instance, Configuration conf, String bindAddress,  int port,
                  int numHandlers, boolean verbose, 
                  SecretManager<? extends TokenIdentifier> secretManager) 
        throws IOException {
      super(bindAddress, port, Invocation.class, numHandlers, conf,
          classNameBase(instance.getClass().getName()), secretManager);
      this.instance = instance;
      this.verbose = verbose;
    }

7.JobTracker()：继续。

    String infoAddr = 
      NetUtils.getServerAddress(conf, "mapred.job.tracker.info.bindAddress",
                                "mapred.job.tracker.info.port",
                                "mapred.job.tracker.http.address");
    InetSocketAddress infoSocAddr = NetUtils.createSocketAddr(infoAddr);
    String infoBindAddress = infoSocAddr.getHostName();
    int tmpInfoPort = infoSocAddr.getPort();
    this.startTime = clock.getTime();
    infoServer = new HttpServer("job", infoBindAddress, tmpInfoPort, 
        tmpInfoPort == 0, conf, aclsManager.getAdminsAcl());
    infoServer.setAttribute("job.tracker", this);
    
    infoServer.addServlet("reducegraph", "/taskgraph", TaskGraphServlet.class);
    infoServer.start();

这里是新建一个infoServer线程用于web服务的，也就是将集群的一些信息显示到wen页面，端口是50030。createInstrumentation()，具体作用不清楚。

8.JobTracker()：继续。

this.dnsToSwitchMapping = ReflectionUtils.newInstance(
        conf.getClass("topology.node.switch.mapping.impl", ScriptBasedMapping.class,
            DNSToSwitchMapping.class), conf);
    this.numTaskCacheLevels = conf.getInt("mapred.task.cache.levels", 
        NetworkTopology.DEFAULT_HOST_LEVEL);
    this.isNodeGroupAware = conf.getBoolean(
            "mapred.jobtracker.nodegroup.aware", false);

新建一个dnsToSwitchMapping用于构造集群的网络拓扑结构。

9.JobTracker()：继续。

plugins = conf.getInstances("mapreduce.jobtracker.plugins",
        ServicePlugin.class);
    for (ServicePlugin p : plugins) {
      try {
        p.start(this);
        LOG.info("Started plug-in " + p + " of type " + p.getClass());
      } catch (Throwable t) {
        LOG.warn("ServicePlugin " + p + " of type " + p.getClass()
            + " could not be started", t);
      }
    }

插件，但是没有实现，应该目前是一个扩展点吧。

10.JobTracker()：最后一句this.initDone.set(conf.getBoolean(JT_INIT_CONFIG_KEY_FOR_TESTS, true))，设置JobTracker初始化完成。

到这里整个JobTracker的构造函数结束了，总结一下，主要对一些重要的对象进行了初始化，有secretManager/aclsManager/taskScheduler/interTrackerServer/infoServicer。接下来回到JobTracker.startTracker()方法。

11.JobTracker.startTracker()：在初始化JobTracker之后，会执行result.taskScheduler.setTaskTrackerManager(result)，将JobTracker赋值给taskScheduler的taskTrackerManager对象。

if(initialize == true) {
        result.setSafeModeInternal(SafeModeAction.SAFEMODE_ENTER);
        result.initializeFilesystem();
        result.setSafeModeInternal(SafeModeAction.SAFEMODE_LEAVE);
        result.initialize();
      }

之前提到调用startTracker时传递的initialize=false，所以这里不进行JobTracker的初始化操作，而是留到offerService()中进行。 到这里main()方法的JobTracker tracker = startTracker(new JobConf())完成，接下来执行tracker.offerService()。

12.JobTracker.offerService()：

this.interTrackerServer.start();

启动interTrackerServer，

public synchronized void start() {
    responder.start();
    listener.start();
    handlers = new Handler[handlerCount];
    
    for (int i = 0; i < handlerCount; i++) {
      handlers[i] = new Handler(i);
      handlers[i].start();
    }
  }

启动responder和listener，同时创建handlerCount个handlers对象并启动，handlerCount前面在JobTracker构造函数中提到，由mapred.job.tracker.handler.count参数决定，默认是10，这里涉及到RPC内部机制。

 setSafeModeInternal(SafeModeAction.SAFEMODE_ENTER);
    initializeFilesystem();
    setSafeModeInternal(SafeModeAction.SAFEMODE_LEAVE);

Initialize the JobTracker FileSystem within safemode。

// Initialize JobTracker
    initialize();

初始化JobTracker。

13.JobTracker.initialize()：

getMROwner().doAs(new PrivilegedExceptionAction<Boolean>() {
      @Override
      public Boolean run() throws Exception {
        JobHistory.init(jtFinal, conf, jtFinal.localMachine, 
            jtFinal.startTime);
        return true;
      }
    });

初始化JobHistory

recoveryManager = new RecoveryManager();

初始化recoveryManager，用于作业恢复管理。

if (conf.getBoolean("mapred.jobtracker.restart.recover", false) 
            && systemDirData != null) {
          for (FileStatus status : systemDirData) {
            try {
              recoveryManager.checkAndAddJob(status);
            } catch (Throwable t) {
              LOG.warn("Failed to add the job " + status.getPath().getName(), 
                       t);
            }
          }
          
          // Check if there are jobs to be recovered
          hasRestarted = recoveryManager.shouldRecover();
          if (hasRestarted) {
            break; // if there is something to recover else clean the sys dir
          }
        }
        LOG.info("Cleaning up the system directory");
        fs.delete(systemDir, true);
        if (FileSystem.mkdirs(fs, systemDir, 
            new FsPermission(SYSTEM_DIR_PERMISSION))) {
          break;
        }

FileStatus[] systemDirData = fs.listStatus(this.systemDir)列出mapred.system.dir参数指定的目录下所有的文件信息，如果mapred.jobtracker.restart.recover=true，默认是false，且systemDir下存在文件时，调用recoveryManager.checkAndAddJob(status)判断是否需要进行作业恢复。

public void checkAndAddJob(FileStatus status) throws IOException {
      String fileName = status.getPath().getName();
      if (isJobNameValid(fileName) && isJobDirValid(JobID.forName(fileName))) {
        recoveryManager.addJobForRecovery(JobID.forName(fileName));
        shouldRecover = true; // enable actual recovery if num-files > 1
      }
    }

根据文件名是否是jobID以及在mapred.system.dir/jobID/下是否存在job信息文件，都存在，则将该jobID添加到recoveryManager队列中，并设置shouldRecover = true，表示需要进行作业恢复。

// Check if there are jobs to be recovered
          hasRestarted = recoveryManager.shouldRecover();
          if (hasRestarted) {
            break; // if there is something to recover else clean the sys dir
          }

判断是否需要进行作业恢复，需要则直接退出，不需要则会执行删除systemDir命令，并重新创建systemDir目录。

fs.delete(systemDir, true);
        if (FileSystem.mkdirs(fs, systemDir, 
            new FsPermission(SYSTEM_DIR_PERMISSION))) {
          break;
        }

之后根据是否需要进行作业恢复，判断是否清楚localDir目录，需要进行恢复则不清楚。

 // Same with 'localDir' except it's always on the local disk.
    if (!hasRestarted) {
      conf.deleteLocalFiles(SUBDIR);
    }

接下来是一些关于JobHistory和infoServer的操作，同时会启动调用jobHistoryServer = new JobHistoryServer(conf, aclsManager, infoServer);jobHistoryServer.start()，启动jobHistoryServer用于查看作业历史信息。

//initializes the job status store
    completedJobStatusStore = new CompletedJobStatusStore(conf, aclsManager);
    
    // Setup HDFS monitoring
    if (this.conf.getBoolean(
        JT_HDFS_MONITOR_ENABLE, DEFAULT_JT_HDFS_MONITOR_THREAD_ENABLE)) {
      hdfsMonitor = new HDFSMonitorThread(this.conf, this, this.fs);
      hdfsMonitor.start();
    }

新建completedJobStatusStore线程，用于将已完成的作业信息保存到hdfs中。具体是在一个job完成之后调用completedJobStatusStore的store()方法将job信息保存到hdfs中，而该线程的run()方法是清除掉已保存到hdfs的job信息文件，该功能的启用由mapred.job.tracker.persist.jobstatus.active参数决定，默认是关闭的。

hdfsMonitor用于监控hdfs是否正常，默认关闭该功能。

到这里initialize()方法完成了。回到offerService()方法。

14.JobTracker.offerService()：recoveryManager.updateRestartCount()，更新恢复次数，具体逻辑：第一次恢复向mapred.system.dir/jobtracker.info文件中写入0，之后每次进行恢复都会对文件中的值进行+1操作。该方法是用来记录恢复次数。taskScheduler.start()，启动taskScheduler，在我上一篇文章中提到，这里不叙述。recoveryManager.recover()，进行作业恢复。

15.RecoveryManager.recover()：

for (JobID jobId : jobsToRecover) {
        LOG.info("Submitting job " + jobId);
        try {
          Path jobInfoFile = getSystemFileForJob(jobId);
          final Path jobTokenFile = getTokenFileForJob(jobId);
          FSDataInputStream in = fs.open(jobInfoFile);
          final JobInfo token = new JobInfo();
          token.readFields(in);
          in.close();
          
          // Read tokens as JT user
          JobConf job = new JobConf();
          final Credentials ts = 
              (jobTokenFile.getFileSystem(job).exists(jobTokenFile)) ?
            Credentials.readTokenStorageFile(jobTokenFile, job) : null;

          // Re-submit job  
          final UserGroupInformation ugi = UserGroupInformation
              .createRemoteUser(token.getUser().toString());
          JobStatus status = ugi.doAs(new PrivilegedExceptionAction<JobStatus>() {
            public JobStatus run() throws IOException, InterruptedException {
              return submitJob(JobID.downgrade(token.getJobID()), token
                  .getJobSubmitDir().toString(), ugi, ts, true);
            }
          });
          if (status == null) {
            LOG.info("Job " + jobId + " was not recovered since it " +
              "disabled recovery on restart (" + JobConf.MAPREDUCE_RECOVER_JOB +
              " set to 'false').");
          } else {
            recovered++;
          }
        } catch (Exception e) {
          LOG.warn("Could not recover job " + jobId, e);
        }
      }

jobsToRecover队列是在jobTracker初始化时添加的，对其中每个作业进行恢复。主要是读取job信息，然后调用JobTracker.submitJob(JobID.downgrade(token.getJobID()), token .getJobSubmitDir().toString(), ugi, ts, true)进行作业在此提交。

// refresh the node list as the recovery manager might have added 
    // disallowed trackers
    refreshHosts();

更新节点信息。

private synchronized void refreshHosts() throws IOException {
    // Reread the config to get mapred.hosts and mapred.hosts.exclude filenames.
    // Update the file names and refresh internal includes and excludes list
    LOG.info("Refreshing hosts information");
    Configuration conf = new Configuration();

    hostsReader.updateFileNames(conf.get("mapred.hosts",""), 
                                conf.get("mapred.hosts.exclude", ""));
    hostsReader.refresh();
    
    Set<String> excludeSet = new HashSet<String>();
    for(Map.Entry<String, TaskTracker> eSet : taskTrackers.entrySet()) {
      String trackerName = eSet.getKey();
      TaskTrackerStatus status = eSet.getValue().getStatus();
      // Check if not include i.e not in host list or in hosts list but excluded
      if (!inHostsList(status) || inExcludedHostsList(status)) {
          excludeSet.add(status.getHost()); // add to rejected trackers
      }
    }
    decommissionNodes(excludeSet);
  }

由hostsReader重新读取mapred.hosts以及mapred.hosts.exclude值，并添加到相应队列中，然后遍历现有的taskTrackers节点，把不在mapred.hosts或者在mapred.hosts.exclude中的节点从hostnameToTaskTracker中移除。

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

这里分别启动expireTrackersThread，retireJobsThread，expireLaunchingTaskThread，completedJobStatusStore线程。

1）expireTrackersThread：该线程用于发现和清除死掉的TaskTracker。主要根据一个TaskTracker自上一次的心跳汇报以来在TASKTRACKER_EXPIRY_INTERVAL时间（由mapred.tasktracker.expiry.interval参数设置，默认是10 * 60 * 1000，单位ms）内未汇报心跳，则将其清除。

2）retireJobsThread：该线程用于清除已完成的作业信息。主要清除那些状态为SUCCEEDED、FAILED、KILLED的job，并且是在RETIRE_JOB_INTERVAL时间间隔（由mapred.jobtracker.retirejob.interval参数设置，默认是24 * 60 * 60 * 1000，单位ms）之前完成；或者用户保存的总完成job数超过MAX_COMPLETE_USER_JOBS_IN_MEMORY（由mapred.jobtracker.completeuserjobs.maximum参数设置，默认是100）也会将已完成的作业进行清除。

3）expireLaunchingTaskThread：该线程用于发现已经分配给某个TaskTracker但一直未汇报信息的任务。主要是根据每个TaskAttemptID启动的时间与当前时间的间隔是否大于TASKTRACKER_EXPIRY_INTERVAL（由mapred.tasktracker.expiry.interval参数设置，默认是10 * 60 * 1000，单位ms）决定。

4）completedJobsStoreThread：该线程用于将已完成的作业信息保存到hdfs中。

至此，JobTracker.offerService()方法完成了。总结下，该方法主要是初始化并启动一些重要线程如上述几个，以及进行作业恢复和启动taskScheduler。

offerService()结束main()也就结束了。这就是整个JobTracker的启动过程，如有错误请指出，谢谢