JobTracker作业启动过程分析

时间：2024-05-01 08:36:33

转自：http://blog.****.net/androidlushangderen/article/details/41356521

在Hadoop中，启动作业运行的方式有很多，可以用命令行格式把打包好后的作业提交还可以，用Hadoop的插件进行应用开发，在这么多的方式中，都会必经过一个流程，作业会以JobInProgress的形式提交到JobTracker中。什么叫JobTracker呢，也许有些人了解Hadoop只知道他的MapReduce计算模型，那个过程只是其中的Task执行的一个具体过程，比较微观上的流程，而JobTrack是一个比较宏观上的东西。涉及到作业的提交的过程。Hadoop遵循的是Master/Slave的架构，也就是主从关系，对应的就是JobTracker/TaskTracker，前者负责资源管理和作业调度，后者主要负责执行由前者分配过来的作业。这样说的话，简单明了。JobTracker里面的执行的过程很多，那就得从开头开始分析，也就是作业最最开始的提交流程开始。后面的分析我会结合MapReduce的代码穿插式的分析，便于大家理解。

其实在作业的提交状态之前，还不会到达JobTacker阶段的，首先是到了MapReduce中一个叫JobClient的类中。也就是说，比如用户通过bin/hadoop jar xxx.jar把打包的jar包上传到系统中时，首先会触发的就是JobClient.。

  1. public RunningJob submitJob(String jobFile) throws FileNotFoundException,
  2. InvalidJobConfException,
  3. IOException {
  4. // Load in the submitted job details
  5. JobConf job = new JobConf(jobFile);
  6. return submitJob(job);
  7. }

之后人家根据配置文件接着调用submitJob()方法

  1. public RunningJob submitJob(JobConf job) throws FileNotFoundException,
  2. IOException {
  3. try {
  4. //又继续调用的是submitJobInternal方法
  5. return submitJobInternal(job);
  6. } catch (InterruptedException ie) {
  7. throw new IOException("interrupted", ie);
  8. } catch (ClassNotFoundException cnfe) {
  9. throw new IOException("class not found", cnfe);
  10. }
  11. }

来到了submitJobInternal的主要方法了

  1. ...
  2. jobCopy = (JobConf)context.getConfiguration();
  3. // Create the splits for the job 为作业创建输入信息
  4. FileSystem fs = submitJobDir.getFileSystem(jobCopy);
  5. LOG.debug("Creating splits at " + fs.makeQualified(submitJobDir));
  6. int maps = writeSplits(context, submitJobDir);
  7. jobCopy.setNumMapTasks(maps);
  8. // write "queue admins of the queue to which job is being submitted"
  9. // to job file.
  10. String queue = jobCopy.getQueueName();
  11. AccessControlList acl = jobSubmitClient.getQueueAdmins(queue);
  12. jobCopy.set(QueueManager.toFullPropertyName(queue,
  13. QueueACL.ADMINISTER_JOBS.getAclName()), acl.getACLString());
  14. // Write job file to JobTracker's fs
  15. FSDataOutputStream out =
  16. FileSystem.create(fs, submitJobFile,
  17. new FsPermission(JobSubmissionFiles.JOB_FILE_PERMISSION));
  18. try {
  19. jobCopy.writeXml(out);
  20. } finally {
  21. out.close();
  22. }
  23. //
  24. // Now, actually submit the job (using the submit name)
  25. //
  26. printTokens(jobId, jobCopy.getCredentials());
  27. //所有信息配置完毕，作业的初始化工作完成，最后将通过RPC方式正式提交作业
  28. status = jobSubmitClient.submitJob(
  29. jobId, submitJobDir.toString(), jobCopy.getCredentials());
  30. JobProfile prof = jobSubmitClient.getJobProfile(jobId);

在这里他会执行一些作业提交之前需要进行的初始化工作，最后会RPC调用远程的提交方法。下面是一个时序图

JobTracker作业启动过程分析

至此我们知道，我们作业已经从本地提交出去了，后面的事情就是JobTracker的事情了，这个时候我们直接会触发的是JobTacker的addJob()方法。

  1. private synchronized JobStatus addJob(JobID jobId, JobInProgress job)
  2. throws IOException {
  3. totalSubmissions++;
  4. synchronized (jobs) {
  5. synchronized (taskScheduler) {
  6. jobs.put(job.getProfile().getJobID(), job);
  7. //观察者模式，会触发每个监听器的方法
  8. for (JobInProgressListener listener : jobInProgressListeners) {
  9. listener.jobAdded(job);
  10. }
  11. }
  12. }
  13. myInstrumentation.submitJob(job.getJobConf(), jobId);
  14. job.getQueueMetrics().submitJob(job.getJobConf(), jobId);
  15. LOG.info("Job " + jobId + " added successfully for user '"
  16. + job.getJobConf().getUser() + "' to queue '"
  17. + job.getJobConf().getQueueName() + "'");
  18. AuditLogger.logSuccess(job.getUser(),
  19. Operation.SUBMIT_JOB.name(), jobId.toString());
  20. return job.getStatus();
  21. }

在这里设置了很多监听器，监听作业的一个情况。那么分析到这里，我们当然也也要顺便学习一下JobTracker的是怎么运行开始的呢。其实JobTracker是一个后台服务程序，他有自己的main方法入口执行地址。上面的英文是这么对此进行描述的：

  1. /**
  2. * Start the JobTracker process.  This is used only for debugging.  As a rule,
  3. * JobTracker should be run as part of the DFS Namenode process.
  4. * JobTracker也是一个后台进程，伴随NameNode进程启动进行，main方法是他的执行入口地址
  5. */
  6. public static void main(String argv[]
  7. ) throws IOException, InterruptedException

上面说的很明白，作为NameNode的附属进程操作，NameNode跟JonTracker一样，全局只有一个，也是Master/Slave的关系对应的是DataNode数据结点。这些是HDFS相关的东西了。

  1. public static void main(String argv[]
  2. ) throws IOException, InterruptedException {
  3. StringUtils.startupShutdownMessage(JobTracker.class, argv, LOG);
  4. try {
  5. if(argv.length == 0) {
  6. //调用startTracker方法开始启动JobTracker
  7. JobTracker tracker = startTracker(new JobConf());
  8. //JobTracker初始化完毕，开启里面的各项线程服务
  9. tracker.offerService();
  10. }
  11. else {
  12. if ("-dumpConfiguration".equals(argv[0]) && argv.length == 1) {
  13. dumpConfiguration(new PrintWriter(System.out));
  14. }
  15. else {
  16. System.out.println("usage: JobTracker [-dumpConfiguration]");
  17. System.exit(-1);
  18. }
  19. }
  20. } catch (Throwable e) {
  21. LOG.fatal(StringUtils.stringifyException(e));
  22. System.exit(-1);
  23. }
  24. }

里面2个主要方法，初始化JobTracker，第二个开启服务方法。首先看startTracker()，最后会执行到new JobTracker()构造函数里面去了：

  1. JobTracker(final JobConf conf, String identifier, Clock clock, QueueManager qm)
  2. throws IOException, InterruptedException {
  3. .....
  4. //初始化安全相关操作
  5. secretManager =
  6. new DelegationTokenSecretManager(secretKeyInterval,
  7. tokenMaxLifetime,
  8. tokenRenewInterval,
  9. DELEGATION_TOKEN_GC_INTERVAL);
  10. secretManager.startThreads();
  11. ......
  12. // Read the hosts/exclude files to restrict access to the jobtracker.
  13. this.hostsReader = new HostsFileReader(conf.get("mapred.hosts", ""),
  14. conf.get("mapred.hosts.exclude", ""));
  15. //初始化ACL访问控制列表
  16. aclsManager = new ACLsManager(conf, new JobACLsManager(conf), queueManager);
  17. LOG.info("Starting jobtracker with owner as " +
  18. getMROwner().getShortUserName());
  19. // Create the scheduler
  20. Class<? extends TaskScheduler> schedulerClass
  21. = conf.getClass("mapred.jobtracker.taskScheduler",
  22. JobQueueTaskScheduler.class, TaskScheduler.class);
  23. //初始化Task任务调度器
  24. taskScheduler = (TaskScheduler) ReflectionUtils.newInstance(schedulerClass, conf);
  25. // Set service-level authorization security policy
  26. if (conf.getBoolean(
  27. ServiceAuthorizationManager.SERVICE_AUTHORIZATION_CONFIG, false)) {
  28. ServiceAuthorizationManager.refresh(conf, new MapReducePolicyProvider());
  29. }
  30. int handlerCount = conf.getInt("mapred.job.tracker.handler.count", 10);
  31. this.interTrackerServer =
  32. RPC.getServer(this, addr.getHostName(), addr.getPort(), handlerCount,
  33. false, conf, secretManager);
  34. if (LOG.isDebugEnabled()) {
  35. Properties p = System.getProperties();
  36. for (Iterator it = p.keySet().iterator(); it.hasNext();) {
  37. String key = (String) it.next();
  38. String val = p.getProperty(key);
  39. LOG.debug("Property '" + key + "' is " + val);
  40. }
  41. }

里面主要干了这么几件事:

1.初始化ACL访问控制列表数据

2.创建TaskSchedule任务调度器

3.得到DPC Server。

4.还有其他一些零零碎碎的操作....

然后第2个方法offService()，主要开启了各项服务;

  1. public void offerService() throws InterruptedException, IOException {
  2. // Prepare for recovery. This is done irrespective of the status of restart
  3. // flag.
  4. while (true) {
  5. try {
  6. recoveryManager.updateRestartCount();
  7. break;
  8. } catch (IOException ioe) {
  9. LOG.warn("Failed to initialize recovery manager. ", ioe);
  10. // wait for some time
  11. Thread.sleep(FS_ACCESS_RETRY_PERIOD);
  12. LOG.warn("Retrying...");
  13. }
  14. }
  15. taskScheduler.start();
  16. .....
  17. this.expireTrackersThread = new Thread(this.expireTrackers,
  18. "expireTrackers");
  19. //启动该线程的主要作用是发现和清理死掉的任务
  20. this.expireTrackersThread.start();
  21. this.retireJobsThread = new Thread(this.retireJobs, "retireJobs");
  22. //启动该线程的作用是清理长时间驻留在内存中且已经执行完的任务
  23. this.retireJobsThread.start();
  24. expireLaunchingTaskThread.start();
  25. if (completedJobStatusStore.isActive()) {
  26. completedJobsStoreThread = new Thread(completedJobStatusStore,
  27. "completedjobsStore-housekeeper");
  28. //该线程的作用是把已经运行完成的任务的信息保存到HDFS中，以便后续的查询
  29. completedJobsStoreThread.start();
  30. }
  31. // start the inter-tracker server once the jt is ready
  32. this.interTrackerServer.start();
  33. synchronized (this) {
  34. state = State.RUNNING;
  35. }
  36. LOG.info("Starting RUNNING");
  37. this.interTrackerServer.join();
  38. LOG.info("Stopped interTrackerServer");
  39. }

主要3大线程在这个方法里被开开启了，expireTrackersThread，retireJobsThread，completedJobsStoreThread，还有1个RPC服务的开启，interTrackerServer.start()，还有细节的操作就不列举出来了。好了JobTraker的close方法的流程刚刚好和以上的操作相反，之前启动过的线程统统关掉。

  1. void close() throws IOException {
  2. //服务停止
  3. if (this.infoServer != null) {
  4. LOG.info("Stopping infoServer");
  5. try {
  6. this.infoServer.stop();
  7. } catch (Exception ex) {
  8. LOG.warn("Exception shutting down JobTracker", ex);
  9. }
  10. }
  11. if (this.interTrackerServer != null) {
  12. LOG.info("Stopping interTrackerServer");
  13. this.interTrackerServer.stop();
  14. }
  15. if (this.expireTrackersThread != null && this.expireTrackersThread.isAlive()) {
  16. LOG.info("Stopping expireTrackers");
  17. //执行线程中断操作
  18. this.expireTrackersThread.interrupt();
  19. try {
  20. //等待线程执行完毕再执行后面的操作
  21. this.expireTrackersThread.join();
  22. } catch (InterruptedException ex) {
  23. ex.printStackTrace();
  24. }
  25. }
  26. if (this.retireJobsThread != null && this.retireJobsThread.isAlive()) {
  27. LOG.info("Stopping retirer");
  28. this.retireJobsThread.interrupt();
  29. try {
  30. this.retireJobsThread.join();
  31. } catch (InterruptedException ex) {
  32. ex.printStackTrace();
  33. }
  34. }
  35. if (taskScheduler != null) {
  36. //调度器的方法终止
  37. taskScheduler.terminate();
  38. }
  39. if (this.expireLaunchingTaskThread != null && this.expireLaunchingTaskThread.isAlive()) {
  40. LOG.info("Stopping expireLaunchingTasks");
  41. this.expireLaunchingTaskThread.interrupt();
  42. try {
  43. this.expireLaunchingTaskThread.join();
  44. } catch (InterruptedException ex) {
  45. ex.printStackTrace();
  46. }
  47. }
  48. if (this.completedJobsStoreThread != null &&
  49. this.completedJobsStoreThread.isAlive()) {
  50. LOG.info("Stopping completedJobsStore thread");
  51. this.completedJobsStoreThread.interrupt();
  52. try {
  53. this.completedJobsStoreThread.join();
  54. } catch (InterruptedException ex) {
  55. ex.printStackTrace();
  56. }
  57. }
  58. if (jobHistoryServer != null) {
  59. LOG.info("Stopping job history server");
  60. try {
  61. jobHistoryServer.shutdown();
  62. } catch (Exception ex) {
  63. LOG.warn("Exception shutting down Job History server", ex);
  64. }
  65. }
  66. DelegationTokenRenewal.close();
  67. LOG.info("stopped all jobtracker services");
  68. return;
  69. }

至此，JobTracker的执行过程总算有了一个了解了吧，不算太难。后面的过程分析。JobTracker是如何把任务进行分解和分配的，从宏观上去理解Hadoop的工作原理。下面是以上过程的一个时序图

JobTracker作业启动过程分析

相关文章