Spark源码之提交任务

首先是Action触发Job的切分，从一个Action开始（file.count()）, RDD中执行job, sc.runjob

  /**
   * Return the number of elements in the RDD.
   */
  def count(): Long = sc.runJob(this, Utils.getIteratorSize _).sum

1、继续深入， 发现在SparkContext中运行runjob

    /**
      * Run a function on a given set of partitions in an RDD and pass the results to the given
      * handler function. This is the main entry point for all actions in Spark.
      */
    def runJob[T, U: ClassTag](
                                  rdd: RDD[T],
                                  func: (TaskContext, Iterator[T]) => U,
                                  partitions: Seq[Int],
                                  resultHandler: (Int, U) => Unit): Unit = {
        if (stopped.get()) {
            throw new IllegalStateException("SparkContext has been shutdown")
        }
        val callSite = getCallSite	
        val cleanedFunc = clean(func)
        logInfo("Starting job: " + callSite.shortForm)
        if (conf.getBoolean("spark.logLineage", false)) {
            logInfo("RDD's recursive dependencies:\n" + rdd.toDebugString)
        }
		//注意dagScheduler
        dagScheduler.runJob(rdd, cleanedFunc, partitions, callSite, resultHandler, localProperties.get)
        progressBar.foreach(_.finishAll())
        rdd.doCheckpoint()
    }

2、DAGScheduler

    /**
      * Run an action job on the given RDD and pass all the results to the resultHandler function as
      * they arrive.
      *
      * @param rdd target RDD to run tasks on
      * @param func a function to run on each partition of the RDD
      * @param partitions set of partitions to run on; some jobs may not want to compute on all
      *   partitions of the target RDD, e.g. for operations like first()
      * @param callSite where in the user program this job was called
      * @param resultHandler callback to pass each result to
      * @param properties scheduler properties to attach to this job, e.g. fair scheduler pool name
      *
      * @throws Exception when the job fails
      */
    def runJob[T, U](
                        rdd: RDD[T],
                        func: (TaskContext, Iterator[T]) => U,
                        partitions: Seq[Int],
                        callSite: CallSite,
                        resultHandler: (Int, U) => Unit,
                        properties: Properties): Unit = {
        val start = System.nanoTime
        //提交job
        val waiter = submitJob(rdd, func, partitions, callSite, resultHandler, properties)
      
        // Note: Do not call Await.ready(future) because that calls `scala.concurrent.blocking`,
        // which causes concurrent SQL executions to fail if a fork-join pool is used. Note that
        // due to idiosyncrasies in Scala, `awaitPermission` is not actually used anywhere so it's
        // safe to pass in null here. For more detail, see SPARK-13747.
        val awaitPermission = null.asInstanceOf[scala.concurrent.CanAwait]
        waiter.completionFuture.ready(Duration.Inf)(awaitPermission)
        waiter.completionFuture.value.get match {
            case scala.util.Success(_) =>
                logInfo("Job %d finished: %s, took %f s".format
                (waiter.jobId, callSite.shortForm, (System.nanoTime - start) / 1e9))
            case scala.util.Failure(exception) =>
                logInfo("Job %d failed: %s, took %f s".format
                (waiter.jobId, callSite.shortForm, (System.nanoTime - start) / 1e9))
                // SPARK-8644: Include user stack trace in exceptions coming from DAGScheduler.
                val callerStackTrace = Thread.currentThread().getStackTrace.tail
                exception.setStackTrace(exception.getStackTrace ++ callerStackTrace)
                throw exception
        }
    }

2.1、submitJob

    /**
      * Submit an action job to the scheduler.
      *
      * @param rdd target RDD to run tasks on
      * @param func a function to run on each partition of the RDD
      * @param partitions set of partitions to run on; some jobs may not want to compute on all
      *   partitions of the target RDD, e.g. for operations like first()
      * @param callSite where in the user program this job was called
      * @param resultHandler callback to pass each result to
      * @param properties scheduler properties to attach to this job, e.g. fair scheduler pool name
      *
      * @return a JobWaiter object that can be used to block until the job finishes executing
      *         or can be used to cancel the job.
      *
      * @throws IllegalArgumentException when partitions ids are illegal
      */
    def submitJob[T, U](
                           rdd: RDD[T],
                           func: (TaskContext, Iterator[T]) => U,
                           partitions: Seq[Int],
                           callSite: CallSite,
                           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
        // ......
       
        assert(partitions.size > 0)
        val func2 = func.asInstanceOf[(TaskContext, Iterator[_]) => _]
        val waiter = new JobWaiter(this, jobId, partitions.size, resultHandler)

		//重点， 通过akka发送job
        eventProcessLoop.post(JobSubmitted(
            jobId, rdd, func2, partitions.toArray, callSite, waiter,
            SerializationUtils.clone(properties)))
        waiter
    }

2.1.1、eventProcessLoop是一个DAGSchedulerEventProcessLoop的对象

主要是 post将job发送到event queue中， 然后线程回去执行， 执行的逻辑在onReceive()中， DAGSchedulerEventProcessLoop的 doOnReceive 通过case match 匹配提交的事件，

    private def doOnReceive(event: DAGSchedulerEvent): Unit = event match {
        
        case JobSubmitted(jobId, rdd, func, partitions, callSite, listener, properties) =>
            dagScheduler.handleJobSubmitted(jobId, rdd, func, partitions, callSite, listener, properties)
		
		//多个case ......
    }

2.1.1.1、handleJobSubmitted

      private[scheduler] def handleJobSubmitted(jobId: Int,
                                              finalRDD: RDD[_],
                                              func: (TaskContext, Iterator[_]) => _,
                                              partitions: Array[Int],
                                              callSite: CallSite,
                                              listener: JobListener,
                                              properties: Properties) {
        var finalStage: ResultStage = null
        //查找最后的Stage , 可能会抛出异常， 当hdfs file不存在
        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 = createResultStage(finalRDD, func, partitions, jobId, callSite)
        } catch {
            case e: Exception =>
                logWarning("Creating new stage failed due to exception - job: " + jobId, e)
                listener.jobFailed(e)
                return
        }

        val job = new ActiveJob(jobId, finalStage, callSite, listener, properties)
        clearCacheLocs()
        //一系列打印
        logInfo("Got job %s (%s) with %d output partitions".format(
            job.jobId, callSite.shortForm, partitions.length))
        logInfo("Final stage: " + finalStage + " (" + finalStage.name + ")")
        logInfo("Parents of final stage: " + finalStage.parents)
        logInfo("Missing parents: " + getMissingParentStages(finalStage))

        val jobSubmissionTime = clock.getTimeMillis()
        jobIdToActiveJob(jobId) = job
        activeJobs += job
        finalStage.setActiveJob(job)
        val stageIds = jobIdToStageIds(jobId).toArray
        val stageInfos = stageIds.flatMap(id => stageIdToStage.get(id).map(_.latestInfo))
       
        listenerBus.post(
            SparkListenerJobStart(job.jobId, jobSubmissionTime, stageInfos, properties))
       
       //提交Stage
        submitStage(finalStage)
    }

2.1.1.1.1、submitStage(stage: Stage), 该方法中递归查找该Stage的父Stage, 并提交

    /** Submits stage, but first recursively submits any missing parents. */
    private def submitStage(stage: Stage) {
        val jobId = activeJobForStage(stage)
        if (jobId.isDefined) {
            logDebug("submitStage(" + stage + ")")
            if (!waitingStages(stage) && !runningStages(stage) && !failedStages(stage)) {
                val missing = getMissingParentStages(stage).sortBy(_.id)
                logDebug("missing: " + missing)
                if (missing.isEmpty) {
                    logInfo("Submitting " + stage + " (" + stage.rdd + "), which has no missing parents")
                    submitMissingTasks(stage, jobId.get)
                } else {
                    for (parent