Hadoop JobTracker提交job源码浅析

妖怪幻

　　上一篇文章说到jobClient提交job的过程，这篇文章是接着上一篇文章继续写的。
　　上一篇说到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;
}


　　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);3.jobTracker.submit()：实例化JobInProgress之后，会根据jobProfile获取job的队列信息，并判断相应的队列是否在运行中，不在则任务失败。然后检查内存情况checkMemoryRequirements(job)，再调用taskScheduler的taskScheduler.checkJobSubmission(job)检查任务提交情况（具体是啥玩意，不太情况）。接下来就是执行status = addJob(jobId, job)，为Job设置listener。　　4.jobTracker.addJob()：前面说过，在初始化jobTracker的时候会实例化taskScheduler，然后调用taskScheduler的start()方法，为jobTracker添加JobListener对象，所以这里的JobInProgressListener对象就是相应的taskScheduler的JobListener，这里为job添加了JobListener。

private synchronized JobStatus addJob(JobID jobId, JobInProgress job)
throws IOException {
totalSubmissions++;
synchronized (jobs) {
synchronized (taskScheduler) {
jobs.put(job.getProfile().getJobID(), job);
for (JobInProgressListener listener : jobInProgressListeners) {
listener.jobAdded(job);
}
}
}
myInstrumentation.submitJob(job.getJobConf(), jobId);
job.getQueueMetrics().submitJob(job.getJobConf(), jobId);
LOG.info("Job " + jobId + " added successfully for user '"
+ job.getJobConf().getUser() + "' to queue '"
+ job.getJobConf().getQueueName() + "'");
AuditLogger.logSuccess(job.getUser(),
Operation.SUBMIT_JOB.name(), jobId.toString());
return job.getStatus();
}到这里整个JobTracker的job提交过程就结束了，中间很多东西没有深入去研究，只是浅显的了解了下，如有错误，请指出，谢谢