spark core源码分析9 从简单例子看action操作，sparkcore

spark core源码分析9 从简单例子看action操作，sparkcore

上一节举例讲解了transformation操作，这一节以reduce为例讲解action操作

首先看submitJob方法，它将我们reduce中写的处理函数随JobSubmitted消息传递出去，因为每个分区都需要调用它进行计算；

而resultHandler是指最后合并的方法，在每个task完成后，需要调用resultHandler将最终结果合并。所以它不需要随JobSubmitted消息传递，而是保存在JobWaiter中

/**
   * Submit a job to the job scheduler and get a JobWaiter object back. The JobWaiter object
   * can be used to block until the the job finishes executing or can be used to cancel the job.
   */
  def submitJob[T, U](
      rdd: RDD[T],
      func: (TaskContext, Iterator[T]) => U,
      partitions: Seq[Int],
      callSite: CallSite,
      allowLocal: Boolean,
      resultHandler: (Int, U) => Unit,
      properties: Properties): JobWaiter[U] = {
    // Check to make sure we are not launching a task on a partition that does not exist.
    val maxPartitions = rdd.partitions.length
    partitions.find(p => p >= maxPartitions || p < 0).foreach { p =>
      throw new IllegalArgumentException(
        "Attempting to access a non-existent partition: " + p + ". " +
          "Total number of partitions: " + maxPartitions)
    }

    val jobId = nextJobId.getAndIncrement()
    if (partitions.size == 0) {
      return new JobWaiter[U](this, jobId, 0, resultHandler)
    }

    assert(partitions.size > 0)
    val func2 = func.asInstanceOf[(TaskContext, Iterator[_]) => _]
    val waiter = new JobWaiter(this, jobId, partitions.size, resultHandler)
    eventProcessLoop.post(JobSubmitted(
      jobId, rdd, func2, partitions.toArray, allowLocal, callSite, waiter,
      SerializationUtils.clone(properties)))
    waiter
  }

finalRDD：触发该action之前的RDD

func：对于每个分区中的元素执行的函数

partitions：分区号Array

listener：这里指JobWaiter

private[scheduler] def handleJobSubmitted(jobId: Int,
      finalRDD: RDD[_],
      func: (TaskContext, Iterator[_]) => _,
      partitions: Array[Int],
      allowLocal: Boolean,
      callSite: CallSite,
      listener: JobListener,
      properties: Properties) {
    var finalStage: ResultStage = null
    try {
      // New stage creation may throw an exception if, for example, jobs are run on a
      // HadoopRDD whose underlying HDFS files have been deleted.
      finalStage = newResultStage(finalRDD, partitions.size, jobId, callSite)
    } catch {
      case e: Exception =>
        logWarning("Creating new stage failed due to exception - job: " + jobId, e)
        listener.jobFailed(e)
        return
    }
    if (finalStage != null) {
      val job = new ActiveJob(jobId, finalStage, func, partitions, callSite, listener, properties)
      clearCacheLocs()
      logInfo("Got job %s (%s) with %d output partitions (allowLocal=%s)".format(
        job.jobId, callSite.shortForm, partitions.length, allowLocal))
      logInfo("Final stage: " + finalStage + "(" + finalStage.name + ")")
      logInfo("Parents of final stage: " + finalStage.parents)
      logInfo("Missing parents: " + getMissingParentStages(finalStage))
      val shouldRunLocally =
        localExecutionEnabled && allowLocal && finalStage.parents.isEmpty && partitions.length == 1
      val jobSubmissionTime = clock.getTimeMillis()
      if (shouldRunLocally) {
        // Compute very short actions like first() or take() with no parent stages locally.
        listenerBus.post(
          SparkListenerJobStart(job.jobId, jobSubmissionTime, Seq.empty, properties))
        runLocally(job)
      } else {
        jobIdToActiveJob(jobId) = job
        activeJobs += job
        finalStage.resultOfJob = Some(job)
        val stageIds = jobIdToStageIds(jobId).toArray
        val stageInfos = stageIds.flatMap(id => stageIdToStage.get(id).map(_.latestInfo))
        listenerBus.post(
          SparkListenerJobStart(job.jobId, jobSubmissionTime, stageInfos, properties))
        submitStage(finalStage)
      }
    }
    submitWaitingStages()
  }

这是一个比较重要的过程，先讲finalStage = newResultStage(finalRDD, partitions.size, jobId, callSite)

这里有一个stage的概念。Task是在集群上运行的基本单位。一个Task负责处理RDD的一个partition。RDD的多个patition会分别由不同的Task去处理，这一组可以同时运行的Task就组成了一个Stage。偷一个官方的图。

以finalRDD为参数构建一个ResultStage

private def newResultStage(
      rdd: RDD[_],
      numTasks: Int,
      jobId: Int,
      callSite: CallSite): ResultStage = {
    val (parentStages: List[Stage], id: Int) = getParentStagesAndId(rdd, jobId)
    val stage: ResultStage = new ResultStage(id, rdd, numTasks, parentStages, jobId, callSite)

    stageIdToStage(id) = stage
    updateJobIdStageIdMaps(jobId, stage)
    stage
  }

解释一下getParentStagesAndId内部的处理逻辑：从finalRDD开始，查找它的所有依赖中的shuffle依赖，如果是普通依赖，则继续往前找，直到找到shuffle依赖为止。这样，就能获取到与finalRDD相邻的所有shuffle依赖。在上图中，即是groupBy和join两个操作产生的依赖。

得到这些shuffle依赖之后，再往前获取整个job所有shuffle依赖，并以shuffle依赖为边界创建ShuffleMapStage，将每个shuffleId注册到mapOutputTracker中，它是跟踪每个shuffleMapStage输出的位置等信息。

在newResultStage方法中，getParentStagesAndId只返回与finalRDD最近的stage

之后再通过父stages，分区数目，stageId，finalRDD，jobId等构建ResultStage。将jobId保存到所有stage的jobIds成员中。一个stage还能有多个jobId？？？

至此，finalStage的建设就完成了。

接着创建了ActiveJob，它只是将那些参数信息封装起来，并有一个成员记录每个partition是否完成。

最后就是调用submitStage将finalStage提交

/** Submits stage, but first recursively submits any missing parents. */
private def submitStage(stage: Stage) {
  val jobId = activeJobForStage(stage)
  if (jobId.isDefined) {
    logDebug("submitStage(" + stage + ")")
    //waitingStages的意思是它还有依赖的父stage还没执行完成时，会先放进这里
    if (!waitingStages(stage) && !runningStages(stage) && !failedStages(stage)) {
      val missing = getMissingParentStages(stage).sortBy(_.id)
      logDebug("missing: " + missing)
      //如果没有父stage未完成，则提交本身的stage
      if (missing.isEmpty) {
        logInfo("Submitting " + stage + " (" + stage.rdd + "), which has no missing parents")
        submitMissingTasks(stage, jobId.get)
      } else {//如果还有未完成的父stage，则递归调用submitStage，先提交父stage，把自己放进waitingStages中
        for (parent <- missing) {
          submitStage(parent)
        }
        waitingStages += stage
      }
    }
  } else {
    abortStage(stage, "No active job for stage " + stage.id)
  }
}

submitMissingTasks代码解析见注释

/** Called when stage's parents are available and we can now do its task. */
private def submitMissingTasks(stage: Stage, jobId: Int) {
  logDebug("submitMissingTasks(