本文分为两个部分：

1. 作业提交流程思维导图
2. 关键函数列表
   

作业提交流程思维导图

点击「链接」查看DataFrame.collect触发的作业提交流程思维导图。

关键函数列表

Dataset.collect

def collect(): Array[T] = withAction("collect", queryExecution)(collectFromPlan)

Dataset.withAction

Dataset.collectFromPlan

触发物理计划的执行，其中 plan 的类型是 SparkPlan

private def collectFromPlan(plan: SparkPlan): Array[T] = {
  val fromRow = resolvedEnc.createDeserializer()
  plan.executeCollect().map(fromRow)
}

Spark 有很多action 函数，比如：

• collect
• count
• show

最终都是通过 collectFromPlan 去创建 Job

SparkPlan.executeCollect

这个函数分为三部：

• getByteArrayRdd函数 将UnsafeRow RDD 转化为 byte array RDD，加速序列化
• 然后调用了 RDD.collect
• 解析 collect 结果，并返回

RDD.collect

Resilient Distributed Dataset (RDD), 是一种不可变、支持分区的数据集合。由于支持分区，该数据集支持并行访问。

class RDD是一个基类，它有很多子类：

• ShuffledRDD：存储shuffle结果数据，parent RDD 是 Java key-value 对
• ShuffledRowRDD：存储shuffle结果数据，parent RDD 是 InternalRow，SparkSQL使用
• MapPartitionsRDD：算子会被应用到 parent RDD 的所有分区
• UnionRDD：存储 union 的结果数据
• 其他 RDD 子类

collect 方法的主要职能是提交 Spark 作业，该功能代理给了 SparkContext 去支持：

SparkContext.runJob

runJob 方法有很多重载，我们只关心最复杂的一个：

从功能上来说，它实现了

• 准备 callSite，以便出问题知道是哪一行代码出错了
• 通过 DAGScheduler.runJob提交作业
• progressBar: 命令行里 stage的进度条显示
• doCheckpoint 将 RDD的中间和最后结果缓存下来

从代码上来说，方法声明如下：

def runJob[T, U: ClassTag](
    rdd: RDD[T],
    func: (TaskContext, Iterator[T]) => U,
    partitions: Seq[Int],
    resultHandler: (Int, U) => Unit): Unit 

它有两个泛型类型参数：

• T: ClassTag 输入RDD的类型
• U: ClassTag 输出数据的类型

参数列表:

• rdd: RDD[T] 指输入RDD类型，比如 RDD[(Long, Array[Byte])]
• func: (TaskContext, Iterator[T]) => U。func会被作用到 rdd的每个分区，返回U
• partitions: Seq[Int]。分区下标列表
• resultHandler: (Int, U)。这是一个回调函数。处理func执行完返回的数据，第一个参数是分区index，第二个是func的返回值

返回值: Unit 表示没有任何返回值

DAGScheduler.runJob

对于 DAGScheduler 而言，Stage是最小的调度单元。它会

• 给Job生成以Stage为调度单位的DAG图
• 追踪RDD和Stage的输出状态，比如哪些已经被物化，并基于这些信息提供一个最优的调度方案
• 提交Stage，以TaskSet的形式提交给 TasksetManager

DAGScheduler 对Job的调度是围绕
DAGSchedulerEventProcessLoop 展开的。这是一个经典的EventLoop使用场景。runJob 方法的执行流程如下：

1. 提交任务本质上是向 EventLoop 发送一个 JobSubmitted 事件
2. 通过一个JobWaiter对象等待结果

在 EventLoop 的另一端，onReceive 接收到 JobSubmitted事件，交给成员函数 handleJobSubmitted 处理该事件。

JobWaiter 内部有一个 Promise 对象，它会不停接收到 taskSucceeded，增加计数，知道成功task的数量等于task的总数量，将promise置为成功。

DAGSchedulerEventProcessLoop.onReceive

onReceive 负责接收各类事件，并分发给特定的 handler 函数处理，具体可以看思维导图或spark代码。

这里我们只看 handleJobSubmitted，它做了五件事情：

• 创建Stage：递归式地创建，先创建parent stage
• 注册Stage
• 创建Job
• 注册Job
• 提交Stage

由于 stage 是一个有向无环图，所以创建和执行都遵循 topological order。

DAGScheduler.createResultStage

在 SparkPlan 对象调用 execute 时，会递归地生成 RDD，从而构成了 RDD Lineage Graph，它是一个有向无环图。那么在 RDD Lineage 上如何切分 stage 呢？

RDD依赖分为宽依赖和窄依赖，代码体现为两个类ShuffleDependency和NarrowDependency。在构建 RDD Lineage时，相邻的两个RDD必须有其中一种依赖关系。Spark通过这种依赖关系划分 Stage。根节点的RDD必须分配到 ResultStage里，而之前所有的Stage，不管有多少级依赖，都是 ShuffleMapStage。

在

DAGScheduler.getShuffleDependenciesAndResourceProfiles

方法中，通过一个栈来记录分配到当前stage中的 RDD（窄依赖中的rdd都会被push到栈里），碰到宽依赖，则加到 shuffleDeps 中。