Hadoop JobTracker提交job源码浅析

397次阅读

共计 9338 个字符，预计需要花费 24 分钟才能阅读完成。

上一篇文章说到 jobClient 提交 job 的过程，这篇文章是接着上一篇文章继续写的 http://www.linuxidc.com/Linux/2013-12/93700.htm。

上一篇说到 jobSubmitClient.submitJob(jobId, submitJobDir.toString(), jobCopy.getCredentials()) 这里，这里就是 jobTracker 进行 job 的提交过程，还有一个 JobSubmissionProtocol 的实现是 LocalJobRunner，这是本地执行的时候使用的，真正集群运行 Job 还是使用的 jobTracker，所以只看 jobTracker 类的 submitJob。

1.jobTracker.submitJob()：第一句就是 checkJobTrackerState() 这个是检查 jobTracker 状态，是否运行中，这里说一句，jobTracker 是在 Hadoop 集群启动的时候启动的，也就是在执行 start-all 或者 start-mapred 的时候启动，启动的时候会调用 JobTracker 的 main 方法，然后在 jps 的时候就可以看见一个 jobTracker 的进程了。下面来看一下 JobTracker.main() 方法。

2.JobTracker.main()：第一句是 JobTracker tracker = startTracker(new JobConf())，这是实例化一个 jobTracke 实例。

3.JobTracker.startTracker()：result = new JobTracker(conf, identifier)，实例化一个 jobTracker 对象，在实例化的时候会做很多事，所以还是进去瞅瞅。

4.JobTracker.JobTracker()：实例化的时候会初始化很多参数，记也记不住，主要看下实例化 taskScheduler 的内容：Class<? extends TaskScheduler> schedulerClass

= conf.getClass(“mapred.jobtracker.taskScheduler”,JobQueueTaskScheduler.class, TaskScheduler.class)；taskScheduler = (TaskScheduler) ReflectionUtils.newInstance(schedulerClass, conf)，这两句就是根据配置文件设置的 taskScheduler 类名，通过反射获得对应的 taskScheduler 对象，在实例化的时候虽然不同的 TaskScheduler 具体操作不一样，但是统一的都会初始化一个 JobListener 对象，这个对象就是后面将要监听 job 的 listener。剩下的内容就不说了。回到 JobTracker.startTracker() 方法。

5.JobTracker.JobTracker()：在实例化 jobTracker 之后，会执行 result.taskScheduler.setTaskTrackerManager(result)，这个就是将 jobTracker 对象设置给 taskScheduler。后面就什么了，现在可以回到 main 方法了

public static JobTracker startTracker(JobConf conf, String identifier, boolean initialize)
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) {
throw e;
} catch (BindException e) {
throw e;
} catch (UnknownHostException e) {
throw e;
} catch (AccessControlException ace) {
// in case of jobtracker not having right access
// bail out
throw ace;
} catch (IOException e) {
LOG.warn(“Error starting tracker: ” +
StringUtils.stringifyException(e));
}
Thread.sleep(1000);
}
if (result != null) {
JobEndNotifier.startNotifier();
MBeans.register(“JobTracker”, “JobTrackerInfo”, result);
if(initialize == true) {
result.setSafeModeInternal(SafeModeAction.SAFEMODE_ENTER);
result.initializeFilesystem();
result.setSafeModeInternal(SafeModeAction.SAFEMODE_LEAVE);
result.initialize();
}
}
return result;
}

更多详情见请继续阅读下一页的精彩内容：http://www.linuxidc.com/Linux/2013-12/93701p2.htm

相关阅读 ：

Ubuntu 13.04 上搭建 Hadoop 环境 http://www.linuxidc.com/Linux/2013-06/86106.htm

Ubuntu 12.10 +Hadoop 1.2.1 版本集群配置 http://www.linuxidc.com/Linux/2013-09/90600.htm

Ubuntu 上搭建 Hadoop 环境（单机模式 + 伪分布模式）http://www.linuxidc.com/Linux/2013-01/77681.htm

Ubuntu 下 Hadoop 环境的配置 http://www.linuxidc.com/Linux/2012-11/74539.htm

单机版搭建 Hadoop 环境图文教程详解 http://www.linuxidc.com/Linux/2012-02/53927.htm

搭建 Hadoop 环境（在 Winodws 环境下用虚拟机虚拟两个 Ubuntu 系统进行搭建）http://www.linuxidc.com/Linux/2011-12/48894.htm

6.JobTracker.main()：在实例化 jobTracker 之后，会调用 tracker.offerService() 方法，之后 main 方法就没什么了，下面看看 tracker.offerService() 这个方法。

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

7.JobTracker.offerService()：这个方法中有一些其他东西，略掉，只看 taskScheduler.start() 这个方法，因为这里只是想分析下 JobTracker 提交 job 的过程，所以省去很多复杂的东西。

8.taskScheduler.start()：这个方法就是启动 TaskScheduler，这个方法不同 taskScheduler 也不同，但是统一的还是会有一个 taskTrackerManager.addJobInProgressListener(jobListener) 这个操作，taskTrackerManager 就是 jobTracker（第 5 步），这句的意思是为 jobTracker 添加 jobListener，用来监听 job 的。这句的内部就是调用 jobTracker 的 jobInProgressListeners 集合的 add(listener) 方法。

到这里可以说看完了整个 JobTracker 的启动过程，虽然很浅显，但是对于后面将要分析的内容，这些就够了。下面来看看 job 的提交过程，也就是 jobTracker 的 submit() 方法。

1.jobTracker.submit()：第一步是 checkSafeMode()，检查是否在安全模式，在安全模式则抛出异常。然后执行 jobInfo = new JobInfo(jobId, new Text(ugi.getShortUserName()),new Path(jobSubmitDir)，生成一个 jobInfo 对象，jobInfo 主要保存 job 的 id,user,jobSubmitDir（也就是 job 的任务目录，上一篇文章提到）。接着是判断 job 是否可被 recovered（job 失败的时候尝试再次执行），如果允许的话 (默认允许)，则将 jobInfo 对象序列化到 job-info 文件中。接着到达最关键的地方，job = new JobInProgress(this, this.conf, jobInfo, 0, ts)，为 job 实例化一个 JobInProgress 对象，这个对象将会对 job 以后的所有情况进行负责，如初始化，执行等。下面看看 JobInProgress 对象的初始化操作。

2.JobInProgress：这里看下将 job.xml 下载到本地的操作。然后就是 job 的队列信息，默认的队列名是 default，Queue queue = this.jobtracker.getQueueManager().getQueue(queueName)，这个主要是根据 Hadoop 所使用的 taskScheduler 有关，具体不研究。剩下的是一些参数的初始化，如 map 的数目，reduce 的数目等。这里还有个设置 job 的优先级的，默认是 normal。this.priority = conf.getJobPriority();this.status.setJobPriority(this.priority); 还有检查 taskLimit 的操作，就是检查 map+reduce 的任务数是否超出 mapred.jobtracker.maxtasks.per.job 设置的值，默认是 -1，就是没有限制的意思。回到 jobTracker.submit() 方法

this.localJobFile = default_conf.getLocalPath(JobTracker.SUBDIR
+”/”+jobId + “.xml”);
Path jobFilePath = JobSubmissionFiles.getJobConfPath(jobSubmitDir);
jobFile = jobFilePath.toString();
fs.copyToLocalFile(jobFilePath, localJobFile);
conf = new JobConf(localJobFile);