概要
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 |
DAGScheduler作用
如上图，DAGScheduler的作用主要有
- compute DAG，执行DAG，得到stage和对应的task，通过TaskScheduler提交到集群，流程大致如下
Spark DAG之SubmitJob
Spark DAG之划分Stage
Spark DAG之SubmitStage
Spark DAG之SubmitTask - preferred locations，就近执行。
根据cache信息和RDD的preferredLocations获取preferred location。 - fault-tolerant，stage级别的容错。
shuffle结束后，reducer读取map的输出，如果读取失败，会触发DAGScheduler重新提交对应的Stage。
示例
以一段代码为例，大致介绍DAGScheduler的作用。
val sc = new SparkContext("local","wordcount")
val data = sc.parallelize(List("a c", "a b", "b c", "b d", "c d"), 2)
val wordcount = data.flatMap(_.split(" ")).map((_, 1)).reduceByKey(_ + _)
wordcount.join(wordcount).collect()
上面代码的逻辑是，先求Wordcount，再做一个join，打印其RDD的依赖关系，如下
如上图，代码中reduceByKey处发生shuffle，共有两处，所以对应的DAG图如下
上图中stage的具体类型以及对应的task类型如下
| stage 0 | stage 1 | stage 2 |
| ShuffleMapStage | ShuffleMapStage | ResultStage |
| ShuffleMapTask | ShuffleMapTask | ResultTask |
最后，DAGScheduler通过TaskScheduler提交ShuffleMapTask和ResultTask到Executor，完成计算。
总结
简单介绍DAGScheduler的作用，以及Stage、Task等概念，并举了一个具体例子，后续结合代码，详细介绍整个流程。
参考：
DAGScheduler
DAGScheduler — Stage-Oriented Scheduler
Spark job submission breakdown
MEMTUNE: Dynamic Memory Management for
In-memory Data Analytic Platforms