学习作业启动

yarn/job_start.md

@@ -136,15 +136,135 @@ dispatcher.getEventHandler().handle(
 
 #### 3、创建job init事件，并且处理
 
+创建init事件，核心代码如下：
+
+```java
+JobEvent initJobEvent = new JobEvent(job.getID(), JobEventType.JOB_INIT);
+jobEventDispatcher.handle(initJobEvent);
+```
+
+事件处理的核心类为InitTransition，核心代码如下：
+
+```java
+public JobStateInternal transition(JobImpl job, JobEvent event) {
+  job.metrics.submittedJob(job);
+  job.metrics.preparingJob(job);
+  // 初始化上下文。
+  if (job.newApiCommitter) {
+    job.jobContext = new JobContextImpl(job.conf,
+        job.oldJobId);
+  } else {
+    job.jobContext = new org.apache.hadoop.mapred.JobContextImpl(
+        job.conf, job.oldJobId);
+  }
+  
+  try {
+    // 初始化token等信息。
+    setup(job);
+    job.fs = job.getFileSystem(job.conf);
+
+    //log to job history
+    JobSubmittedEvent jse = new JobSubmittedEvent(job.oldJobId,
+          job.conf.get(MRJobConfig.JOB_NAME, "test"), 
+        job.conf.get(MRJobConfig.USER_NAME, "mapred"),
+        job.appSubmitTime,
+        job.remoteJobConfFile.toString(),
+        job.jobACLs, job.queueName,
+        job.conf.get(MRJobConfig.WORKFLOW_ID, ""),
+        job.conf.get(MRJobConfig.WORKFLOW_NAME, ""),
+        job.conf.get(MRJobConfig.WORKFLOW_NODE_NAME, ""),
+        getWorkflowAdjacencies(job.conf),
+        job.conf.get(MRJobConfig.WORKFLOW_TAGS, ""), job.conf);
+    job.eventHandler.handle(new JobHistoryEvent(job.jobId, jse));
+    //TODO JH Verify jobACLs, UserName via UGI?
+    // 初始化并行度等信息。
+    TaskSplitMetaInfo[] taskSplitMetaInfo = createSplits(job, job.jobId);
+    job.numMapTasks = taskSplitMetaInfo.length;
+    job.numReduceTasks = job.conf.getInt(MRJobConfig.NUM_REDUCES, 0);
+
+    if (job.numMapTasks == 0 && job.numReduceTasks == 0) {
+      job.addDiagnostic("No of maps and reduces are 0 " + job.jobId);
+    } else if (job.numMapTasks == 0) {
+      job.reduceWeight = 0.9f;
+    } else if (job.numReduceTasks == 0) {
+      job.mapWeight = 0.9f;
+    } else {
+      job.mapWeight = job.reduceWeight = 0.45f;
+    }
+
+    checkTaskLimits();
+    
+   // 加载其他参数，具体代码省略。。
+
+    cleanupSharedCacheUploadPolicies(job.conf);
+
+    // create the Tasks but don't start them yet，， 创建map task
+    createMapTasks(job, inputLength, taskSplitMetaInfo);
+    // 创建reduce tasks
+    createReduceTasks(job);
+
+    job.metrics.endPreparingJob(job);
+    return JobStateInternal.INITED;
+  } catch (Exception e) {
+    LOG.warn("Job init failed", e);
+    job.metrics.endPreparingJob(job);
+    job.addDiagnostic("Job init failed : "
+        + StringUtils.stringifyException(e));
+    // Leave job in the NEW state. The MR AM will detect that the state is
+    // not INITED and send a JOB_INIT_FAILED event.
+    return JobStateInternal.NEW;
+  }
+}
+```
 
 
 
+#### 4、检查初始化结果并且启动作业
 
+当init成功时，handler返回的结果是JobStateInternal.INITED；如果是失败了则返回的结果是JobStateInternal.NEW。
 
+对于初始化失败的作业会触发JobEventType.JOB_INIT_FAILED事件。
 
+对于初始化成功的作业会调用函数startJobs，继续启动作业。触发
 
+```java
+protected void startJobs() {
+  /** create a job-start event to get this ball rolling */
+  JobEvent startJobEvent = new JobStartEvent(job.getID(),
+      recoveredJobStartTime);
+  /** send the job-start event. this triggers the job execution. */
+  dispatcher.getEventHandler().handle(startJobEvent);
+}
+```
 
+核心处理逻辑如下，主要是触发了几个事件：
 
+- JobHistoryEvent：
+- JobInfoChangeEvent：
+- CommitterJobSetupEvent：
 
+```java
+public void transition(JobImpl job, JobEvent event) {
+  JobStartEvent jse = (JobStartEvent) event;
+  if (jse.getRecoveredJobStartTime() != -1L) {
+    job.startTime = jse.getRecoveredJobStartTime();
+  } else {
+    job.startTime = job.clock.getTime();
+  }
+  JobInitedEvent jie =
+    new JobInitedEvent(job.oldJobId,
+         job.startTime,
+         job.numMapTasks, job.numReduceTasks,
+         job.getState().toString(),
+         job.isUber());
+  job.eventHandler.handle(new JobHistoryEvent(job.jobId, jie));
+  JobInfoChangeEvent jice = new JobInfoChangeEvent(job.oldJobId,
+      job.appSubmitTime, job.startTime);
+  job.eventHandler.handle(new JobHistoryEvent(job.jobId, jice));
+  job.metrics.runningJob(job);
 
+  job.eventHandler.handle(new CommitterJobSetupEvent(
+          job.jobId, job.jobContext));
+}
+```