Spark2.4.0源码分析之WorldCount 触发作业提交(二)

青旅半醒 2022-02-13 01:55 225阅读 0赞

# Spark2.4.0源码分析之WorldCount 触发作业提交(二) #

## 更多资源 ##

*  github: [https://github.com/opensourceteams/spark-scala-maven-2.4.0][https_github.com_opensourceteams_spark-scala-maven-2.4.0]

## 主要内容描述 ##

*  Final RDD作为参数，通过RDD.collect()函数触发作业提交

## 时序图 ##

*  [/images/20220213/8b6663c4518d4877b9baefd72a66a8a1.png][https_github.com_opensourceteams_spark-scala-maven-2.4.0_blob_master_md_image_example_spark-sql-dataset_worldCount_worldcount.trigger.job.submit.jpg]  
    ![https_github.com_opensourceteams_spark-scala-maven-2.4.0_blob_master_md_image_example_spark-sql-dataset_worldCount_worldcount.trigger.job.submit.jpg][]

## 源码分析 ##

### SparkPlan.executeCollect ###

*  val byteArrayRdd = getByteArrayRdd()得到MapPartitionsRDD \[9\],即通过Spark执行计划转化为Final RDD
 *  调用RDD.collect()函数对RDD进行处理

/**
       * Runs this query returning the result as an array.
       */
      def executeCollect(): Array[InternalRow] = {
        val byteArrayRdd = getByteArrayRdd()
    
        val results = ArrayBuffer[InternalRow]()
        byteArrayRdd.collect().foreach { countAndBytes =>
          decodeUnsafeRows(countAndBytes._2).foreach(results.+=)
        }
        results.toArray
      }

### RDD.collect ###

*  调用函数 SparkContext.runJob()

/**
       * Return an array that contains all of the elements in this RDD.
       *
       * @note This method should only be used if the resulting array is expected to be small, as
       * all the data is loaded into the driver's memory.
       */
      def collect(): Array[T] = withScope {
        val results = sc.runJob(this, (iter: Iterator[T]) => iter.toArray)
        Array.concat(results: _*)
      }

### SparkContext.runJob ###

*  调用同名函数SparkContext.runJob

/**
       * Run a job on all partitions in an RDD and return the results in an array.
       *
       * @param rdd target RDD to run tasks on
       * @param func a function to run on each partition of the RDD
       * @return in-memory collection with a result of the job (each collection element will contain
       * a result from one partition)
       */
      def runJob[T, U: ClassTag](rdd: RDD[T], func: Iterator[T] => U): Array[U] = {
        runJob(rdd, func, 0 until rdd.partitions.length)
      }

### SparkContext.runJob ###

*  调用同名函数SparkContext.runJob

/**
       * Run a function on a given set of partitions in an RDD and return the results as an array.
       *
       * @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()`
       * @return in-memory collection with a result of the job (each collection element will contain
       * a result from one partition)
       */
      def runJob[T, U: ClassTag](
          rdd: RDD[T],
          func: Iterator[T] => U,
          partitions: Seq[Int]): Array[U] = {
        val cleanedFunc = clean(func)
        runJob(rdd, (ctx: TaskContext, it: Iterator[T]) => cleanedFunc(it), partitions)
      }

### SparkContext.runJob ###

*  调用同名函数SparkContext.runJob

/**
       * Run a function on a given set of partitions in an RDD and return the results as an array.
       * The function that is run against each partition additionally takes `TaskContext` argument.
       *
       * @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()`
       * @return in-memory collection with a result of the job (each collection element will contain
       * a result from one partition)
       */
      def runJob[T, U: ClassTag](
          rdd: RDD[T],
          func: (TaskContext, Iterator[T]) => U,
          partitions: Seq[Int]): Array[U] = {
        val results = new Array[U](partitions.size)
        runJob[T, U](rdd, func, partitions, (index, res) => results(index) = res)
        results
      }

### SparkContext.runJob ###

*  调用DAGScheduler.runJob()函数来处理rdd

/**
       * 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.
       *
       * @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 resultHandler callback to pass each result to
       */
      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.runJob(rdd, cleanedFunc, partitions, callSite, resultHandler, localProperties.get)
        progressBar.foreach(_.finishAll())
        rdd.doCheckpoint()
      }

### DAGScheduler.runJob() ###

*  调用函数DAGScheduler.submitJob()来处理RDD

/**
       * 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
       *
       * @note 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
        val waiter = submitJob(rdd, func, partitions, callSite, resultHandler, properties)
        ThreadUtils.awaitReady(waiter.completionFuture, Duration.Inf)
        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
        }
      }

### DAGScheduler.submitJob() ###

*  调用事件循环器，把JobSubmitted事件提交给事件队列中
 *  事件处理器会对作业提交这一事件进行处理
 *  本文只对触发作业进行分析，后续操作请查看关联文档

/**
       * 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.
        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 immediately if the job is running 0 tasks
          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, callSite, waiter,
          SerializationUtils.clone(properties)))
        waiter
      }

end

[https_github.com_opensourceteams_spark-scala-maven-2.4.0]: https://github.com/opensourceteams/spark-scala-maven-2.4.0
[https_github.com_opensourceteams_spark-scala-maven-2.4.0_blob_master_md_image_example_spark-sql-dataset_worldCount_worldcount.trigger.job.submit.jpg]: /images/20220213/8b6663c4518d4877b9baefd72a66a8a1.png