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

更多资源

• github:

主要内容描述

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

时序图

源码分析

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