DAG 概述
DAG，有向无环图，Directed Acyclic Graph的缩写，常用于建模。Spark中使用DAG对RDD的关系进行建模，描述了RDD的依赖关系，这种关系也被称之为lineage，RDD的依赖关系使用Dependency维护，参考Spark RDD之Dependency，DAG在Spark中的对应的实现为DAGScheduler。
介绍DAGScheduler中的一些概念，有助于理解后续流程。
| 名词 | 解释 |
|---|---|
| Job | 调用RDD的一个action，如count，即触发一个Job，spark中对应实现为ActiveJob，DAGScheduler中使用集合activeJobs和jobIdToActiveJob维护Job |
| Stage | 代表一个Job的DAG，会在发生shuffle处被切分，切分后每一个部分即为一个Stage，Stage实现分为ShuffleMapStage和ResultStage，一个Job切分的结果是0个或多个ShuffleMapStage加一个ResultStage |
| Task | 最终被发送到Executor执行的任务，和stage的ShuffleMapStage和ResultStage对应，其实现分为ShuffleMapTask和ResultTask |
Submit Job
在介绍 Submit Job 前，需要先引入一个概念：EventLoop
EventLoop
DAGScheduler 使用 EventLoop(LinkedBlockingDeque)异步处理Job的流程，程序由同步改为异步是优化并发，提升性能的常见手段，在spark中使用的非常多。
异步处理借助于 EventLoop 实现，EventLoop内部维护了LinkedBlockingDeque，LinkedBlockingDeque是基于链表实现的双端阻塞队列，定义如下：
private[spark] abstract class EventLoop[E](name: String) extends Logging {
private val eventQueue: BlockingQueue[E] = new LinkedBlockingDeque[E]()
private val stopped = new AtomicBoolean(false)
private val eventThread = new Thread(name) {
override def run(): Unit = {
while (!stopped.get) {
val event = eventQueue.take()
onReceive(event)
}
}
}
//抽象方法，需要在子类实现
protected def onReceive(event: E): Unit
...
}
DAGSchedulerEvent
DAGScheduler对事件进行了分类。
事件的父类为 DAGSchedulerEvent，也是EventLoop中存储的类型。
具体子类类型为：
上面的事件基本能够见名知义，下面 Submit Job 部分会涉及到第一个事件JobSubmitted。
DAGSchedulerEventProcessLoop
EventLoop的实现类，主要为抽象方法onReceive的实现，处理各种不同DAGSchedulerEvent。如下：
DAGSchedulerEventProcessLoop：
private[scheduler] class DAGSchedulerEventProcessLoop(dagScheduler: DAGScheduler)
extends EventLoop[DAGSchedulerEvent]("dag-scheduler-event-loop") with Logging {
override def onReceive(event: DAGSchedulerEvent): Unit = {
doOnReceive(event)
}
}
DAGSchedulerEventProcessLoop 继承自 EventLoop，并重写了 onReceive 方法，看下 doOnReceive：
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 MapStageSubmitted(jobId, dependency, callSite, listener, properties) =>
dagScheduler.handleMapStageSubmitted(jobId, dependency, callSite, listener, properties)
...
}
onReceive方法调用doOnReceive，doOnReceive中根据事件的类型，调用DAGScheduler的不同方法处理。如上面提到的JobSubmitted事件，交给DAGScheduler的handleJobSubmitted方法处理。
SubmitJob
Spark 任务调度之Driver send Task 我们介绍了SparkContext的runJob方法调用DAGScheduler的runJob方法，把RDD交给DAGScheduler处理。
查看DAGScheduler 中 runJob方法中的部分代码：
submitJob 方法部分代码：
eventProcessLoop.post(JobSubmitted(
jobId, rdd, func2, partitions.toArray, callSite, waiter,
SerializationUtils.clone(properties)))
调用 DAGSchedulerEventProcessLoop 的 post方法，post是在父类 EventLoop 实现：
def post(event: E): Unit = {
eventQueue.put(event)
}
把事件添加到队列 eventQueue，上面代码中提到，EventLoop 中创建了线程，并在线程中 只要while (!stopped.get)，就会从队列中取event：val event = eventQueue.take()，并回调 onReceive(event)，最终在 DAGSchedulerEventProcessLoop 中消费事件。
总结
介绍了EventLoop的概念及DAGScheduler使用EventLoop异步处理Job的流程，但是EventLoop中维护的LinkedBlockingDeque并没有指定容量，默认容量为Integer.MAX_VALUE，如果eventThread消费不及时，有OOM的风险，最后 DAGScheduler 消费 JobSubmitted 事件的流程大致如下
- ①-④流程，提交 JobSubmitted 事件到 LinkedBlockingDeque。
- (1)-(4)流程，eventThread循环消费 LinkedBlockingDeque，最终将JobSubmitted事件交给 DAGScheduler 的 handleJobSubmitted 方法处理。
