当触发一个RDD的action后，以count为例，调用关系如下：

• 复制代码
  
  

其中步骤五的DAGSchedulerEventProcessActor是DAGScheduler 的与外部交互的接口代理，DAGScheduler在创建时会创建名字为eventProcessActor的actor。这个actor的作用看它的实现就一目了然了：

• 复制代码
  
  

总结一下org.apache.spark.scheduler.DAGSchedulerEventProcessActor的作用：可以把他理解成DAGScheduler的对外的功能接口。它对外隐藏了自己内部实现的细节，也更易于理解其逻辑；也降低了维护成本，将DAGScheduler的比较复杂功能接口化。

handleJobSubmitted

---

    org.apache.spark.scheduler.DAGScheduler#handleJobSubmitted首先会根据RDD创建finalStage。

finalStage，顾名思义，就是最后的那个Stage。然后创建job，最后提交。提交的job如果满足一下条件，那么它将以本地模式运行：

     1）spark.localExecution.enabled设置为true  并且 2）用户程序显式指定可以本地运行 并且 3）finalStage的没有父Stage 并且 4）仅有一个partition

     3）和 4）的话主要为了任务可以快速执行；如果有多个stage或者多个partition的话，本地运行可能会因为本机的计算资源的问题而影响任务的计算速度。

      要理解什么是Stage，首先要搞明白什么是Task。Task是在集群上运行的基本单位。一个Task负责处理RDD的一个partition。RDD的多个patition会分别由不同的Task去处理。当然了这些Task的处理逻辑完全是一致的。这一组Task就组成了一个Stage。有两种Task：

• org.apache.spark.scheduler.ShuffleMapTask
  
• org.apache.spark.scheduler.ResultTask
  
  

      ShuffleMapTask根据Task的partitioner将计算结果放到不同的bucket中。而ResultTask将计算结果发送回Driver Application。一个Job包含了多个Stage，而Stage是由一组完全相同的Task组成的。最后的Stage包含了一组ResultTask。

      在用户触发了一个action后，比如count，collect，SparkContext会通过runJob的函数开始进行任务提交。最后会通过DAG的event processor 传递到DAGScheduler本身的handleJobSubmitted，它首先会划分Stage，提交Stage，提交Task。至此，Task就开始在运行在集群上了。

     一个Stage的开始就是从外部存储或者shuffle结果中读取数据；一个Stage的结束就是由于发生shuffle或者生成结果时。

创建finalStage

---

handleJobSubmitted 通过调用newStage来创建finalStage：

• finalStage = newStage(finalRDD, partitions.size, None, jobId, callSite)
  
  

创建一个result stage，或者说finalStage，是通过调用org.apache.spark.scheduler.DAGScheduler#newStage完成的；而创建一个shuffle stage，需要通过调用org.apache.spark.scheduler.DAGScheduler#newOrUsedStage。

复制代码

对于result 的final stage来说，传入的shuffleDep是None。

我们知道，RDD通过org.apache.spark.rdd.RDD#getDependencies可以获得它依赖的parent RDD。而Stage也可能会有parent Stage。看一个RDD论文的Stage划分吧：

      一个stage的边界，输入是外部的存储或者一个stage shuffle的结果；输入则是Job的结果（result task对应的stage）或者shuffle的结果。

      上图的话stage3的输入则是RDD A和RDD F shuffle的结果。而A和F由于到B和G需要shuffle，因此需要划分到不同的stage。

从源码实现的角度来看，通过触发action也就是最后一个RDD创建final stage（上图的stage 3），我们注意到new Stage的第五个参数就是该Stage的parent Stage：通过rdd和job id获取：

• 复制代码
  
  

生成了finalStage后，就需要提交Stage了。

• 复制代码
  
  

DAGScheduler将Stage划分完成后，提交实际上是通过把Stage转换为TaskSet，然后通过TaskScheduler将计算任务最终提交到集群。其所在的位置如下图所示。

接下来，将分析Stage是如何转换为TaskSet，并最终提交到Executor去运行的。