概要

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维护JobStage代表一个Job的DAG，会在发生shuffle处被切分，切分后每一个部分即为一个Stage，Stage实现分为ShuffleMapStage和ResultStage，一个Job切分的结果是0个或多个ShuffleMapStage加一个ResultStageTask最终被发送到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之SubmitTaskpreferred 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 0stage 1stage 2ShuffleMapStageShuffleMapStageResultStageShuffleMapTaskShuffleMapTaskResultTask

最后，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