一个Spark job的生命历程

dagScheduler.runJob //（1）
--> submitJob ( eventProcessLoop.post(JobSubmitted,***) //（2）
    --> eventProcessLoop //（3）
        --> onReceive(event: DAGSchedulerEvent) //（4）
            --> doOnReceive(event: DAGSchedulerEvent) //（5）
                --> case JobSubmitted //（6）
                    --> dagScheduler.handleJobSubmitted //（7）
                        --> finalStage =createResultStage(finalRDD, func, partitions, jobId, callSite) //（8）    
                        --> job = new ActiveJob(jobId, finalStage, callSite, listener, properties) //（9）
                        --> jobIdToActiveJob(jobId) = job //（10）
                        --> activeJobs += job //（11）
                        --> finalStage.setActiveJob(job) //（12）
                        --> stageIds = jobIdToStageIds(jobId).toArray //（13）
                        --> stageInfos = stageIds.flatMap(id => stageIdToStage.get(id).map(_.latestInfo)) //（14）
                        --> listenerBus.post(SparkListenerJobStart(job.jobId, jobSubmissionTime, stageInfos, properties)) //（15）
                        --> submitStage(finalStage) //（16）
                            --> getMissingParentStages(stage).sortBy(_.id) //（17）
                                --> finalStage = getOrCreateShuffleMapStage(dependency, jobId) //（18）
                                    --> createShuffleMapStage(dep, firstJobId) //（19）
                                        -->stage = new ShuffleMapStage(id, rdd, numTasks, parents, jobId, rdd.creationSite, shuffleDep)
                                --> job = new ActiveJob(jobId, finalStage, callSite, listener, properties) //（20）                    
                                --> submitStage(finalStage)  //（21）//划分和提交stage算法精髓
                                    --> submitMissingTasks(stage, jobId.get) //（22）
                                        --> submitWaitingChildStages(stage) //(23)
                                --> markMapStageJobAsFinished(job, mapOutputTracker.getStatistics(dependency))  //（24）

/**
* 获取某个stage的父stage
* 对于一个stage，如果它的最后一个RDD的所有依赖都是窄依赖，将不会创建新的stage
* 如果其RDD会依赖某个RDD，用宽依赖的RDD创建一个新的stage，并立即返回这个stage
* @type {[type]}
*/
private def getMissingParentStages(stage: Stage): List[Stage] = {
    val missing = new HashSet[Stage]
    val visited = new HashSet[RDD[_]]
    val waitingForVisit = new Stack[RDD[_]]
    
    def visit(rdd: RDD[_]) {
      if (!visited(rdd)) {
        visited += rdd
        val rddHasUncachedPartitions = getCacheLocs(rdd).contains(Nil)
        if (rddHasUncachedPartitions) {
            //遍历RDD的依赖，对于每种具有shuffle的操作，如reduceByKey，groupByKey，countByKey，底层对应了3个RDD：
            //Map
          for (dep <- rdd.dependencies) {
            dep match {
                //如果是宽依赖
              case shufDep: ShuffleDependency[_, _, _] =>
                  //使用宽依赖的RDD创建一个 ShuffleMapStage，并且将isShuffleMap 设置为true，
                  //默认最后一个stage不是shuffle不是ShuffleMapStage，但是finalstage之前所有的stage都是ShuffleMapStage
                val mapStage = getOrCreateShuffleMapStage(shufDep, stage.firstJobId)
                if (!mapStage.isAvailable) {
                  missing += mapStage
                }
              
                //如果是窄依赖
              case narrowDep: NarrowDependency[_] =>
              //将依赖的RDD放入栈中
                waitingForVisit.push(narrowDep.rdd)
            }
          }
        }
      }
    }
    //
    waitingForVisit.push(stage.rdd)
    while (waitingForVisit.nonEmpty) {
    //
      visit(waitingForVisit.pop())
    }
    missing.toList
  }

override def submitTasks(taskSet: TaskSet) {
    val tasks = taskSet.tasks　//获取ttaskSet的task列表
    logInfo("Adding task set " + taskSet.id + " with " + tasks.length + " tasks")
    this.synchronized {
      //每个taskSet都会创建一个manager，用于管理每个taskSet，并设定最大失败次数 maxTaskFailures
      val manager = createTaskSetManager(taskSet, maxTaskFailures)
      val stage = taskSet.stageId
      //尝试连接task，如果task失败，会负责重试spark,直到超过重试次数,并且会通知延迟调度
      val stageTaskSets =
        taskSetsByStageIdAndAttempt.getOrElseUpdate(stage, new HashMap[Int, TaskSetManager])
      //通过 manager 获得活着的taskSet
      stageTaskSets(taskSet.stageAttemptId) = manager
      val conflictingTaskSet = stageTaskSets.exists { case (_, ts) =>
        ts.taskSet != taskSet && !ts.isZombie
      }
      if (conflictingTaskSet) {
        throw new IllegalStateException(s"more than one active taskSet for stage $stage:" +
          s" ${stageTaskSets.toSeq.map{_._2.taskSet.id}.mkString(",")}")
      }
      //利用已选择的调度器schedulableBuilder，把一个taskSet的manager加入调度管理池中
      /*
      def initialize(backend: SchedulerBackend) {
        this.backend = backend
        schedulableBuilder = {
          schedulingMode match {
            case SchedulingMode.FIFO =>
              new FIFOSchedulableBuilder(rootPool)
            case SchedulingMode.FAIR =>
              new FairSchedulableBuilder(rootPool, conf)
            case _ =>
              throw new IllegalArgumentException(s"Unsupported $SCHEDULER_MODE_PROPERTY: " +
              s"$schedulingMode")
          }
        }
        schedulableBuilder.buildPools()
      }*/
      schedulableBuilder.addTaskSetManager(manager, manager.taskSet.properties)
      
      if (!isLocal && !hasReceivedTask) {
        starvationTimer.scheduleAtFixedRate(new TimerTask() {
          override def run() {
            if (!hasLaunchedTask) {
              logWarning("Initial job has not accepted any resources; " +
                "check your cluster UI to ensure that workers are registered " +
                "and have sufficient resources")
            } else {
              this.cancel()
            }
          }
        }, STARVATION_TIMEOUT_MS, STARVATION_TIMEOUT_MS)
      }
      hasReceivedTask = true
    }
    /**
      * 创建 taskScheduler 的时候,就是为 taskSchedulerImpl 创建一个 SparkDeploySchedulerBackend .
      * 它负责创建AppClient,向master注册Application
      */
    backend.reviveOffers()
  }