spark源码学习（三）：job的提交以及runJob函数的分析

spark源码学习：sparkContext的初始化分析（三）

上一篇仅仅谈及了taskScheduler和schedulerBackend相关的代码。这篇blog记录一下和DAGScheduled相关的代码。DAGScheduled是底层的面向stage调度的划分，还会去跟踪哪些RDD和stage的输出已经物化，并且去找到一个最优的执行方案，也就是指这种调度花费的资源调度，然后把stage作为taskSet向taskScheduler去提交。另外，还可以去决定这些任务在哪些locations上去执行会更加合适，也可以去对那些已经提交的stage，但是输出结果却已经丢失了，这种错误DAGScheduled也可以搞定。

这里的对于一个job，DAGScheduled把job划分成几个相互关联stage，stage又包含若干个独立的同时运行的task任务。这里job，stage，task的提交的顺序确实恰巧相反的。大致是首先创建的finalStage，然后根据fialStage创建activeJob,之后就是根据shuffle的类型去倒推得到出stage。然后再正向去通过submitMissingTask来创建task，task只有两种task，shuffleMapTask和resultTask，只有最后的那些并发的task是result-Task。之所以能知道这些依赖的属性，是宽依赖和窄依赖，是每个rdd在创建的时候就知道的。如下图所示：



上面的图片显示了很多的信息，例如runJob,dagScheduler.runJob( )函数，submitStage( )函数，visit( )到底都是些啥东西呢？并且runJob( )函数是啥时候提交的呢？下面我们从一个简单的函数来开始看看具体的提交过程。



哈哈，sc.runJob( )函数出现了，点击进入之后进入sc.runJob( )函数。



我去，又是一个runJob函数，false就是代表时候支持本地运行模式。func就是一个函数而已，后面会用到。再看看里面又有一个runJob函数，又多了两个参数，一个对于rdd分片的迭代，另一个是bollean。再进去看看：



这个runJob函数里面又有一个runJob函数，里面的runJob函数稍微的改变一下函数的模式，添加了一个参数TaskContext，就是任务的上下文，可以在任务执行的时候可读并且共享的上下文。那就继续进入吧。。。



这里的(index,res)=>result(index)=res;就是对rdd的分片逐个进行计算，并且保存results，是一个Array(分片大小)

这个里面的函数就是添加一个results，那就继续下去看看这个runJob函数吧。



哈哈，在dagScheduler.runJob( )运行结束之后，还进做了一个检查点机制。callSite主要就是得到用户栈的栈顶和属于spark核心类栈的栈底。下面进入里面的runJob函数：



进入这个submitJob函数，看看吧：



eventProcessLoop.post(JobSubmitted)函数是一个核心，核心啊，哈哈。我们来一一描述：

(1) eventProcessLoop:它是一个DAGSchedulerEventProcessLoop，而这个类就是一个EventLoop[DAGSch-edulerEvent]。EventLoop是从caller接受事件并且处理，它将开启一个唯一的线程event，由于event会无限制的增长，所以必须要有一个post函数来处理事件以防止内存泄漏。继而DAGSchedulerEvent是什么？他是一个可以被DAGSchedulerEvent可以处理的时间，它里面有很多的case
class 类例如JobSubmitted ,StageCancelled等等事件。

(2)post函数就是会把时间Event放到队列事件中，事件之后会被线程处理。

(3)jobSubmitted也就是一个可以被处理的事件。

下面我们就看看在post这个事件到队列中之后，onReceive接收这和提交的事件，会怎样除了呢？看看code吧：



这里才会真正的进入依赖性的分析，下面的核心函数讨论就是handleJobSubmitted，newResultStage ,ActiveJob ,submitWaitingStages ,submitStage,getMissingParentStages,submitMissingTasks.哈哈，感觉很多吧，这些都是和stage划分的核心方法，下篇blog我们再来讨论。