本文共 1770 字，大约阅读时间需要 5 分钟。

概要

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 ---------------------  原文：https://blog.csdn.net/u011564172/article/details/70172060