Spark2.0源码学习-Job提交与Task的拆分

【Spark2.0源码学习】-9.Job提交与Task的拆分

      在前面的章节Client的加载中，Spark的DriverRunner已开始执行用户任务类（比如：org.apache.spark.examples.SparkPi），下面我们开始针对于用户任务类（或者任务代码）进行分析
 
一、整体预览
          基于上篇图做了扩展，增加任务执行的相关交互
     


Code：指的用户编写的代码
RDD：弹性分布式数据集，用户编码根据SparkContext与RDD的api能够很好的将Code转化为RDD数据结构（下文将做转化细节介绍）
DAGScheduler：有向无环图调度器，将RDD封装为JobSubmitted对象存入EventLoop（实现类DAGSchedulerEventProcessLoop）队列中
EventLoop： 定时扫描未处理JobSubmitted对象，将JobSubmitted对象提交给DAGScheduler
DAGScheduler：针对于JobSubmitted进行处理，最终将RDD转化为执行TaskSet,并将TaskSet提交至TaskScheduler
TaskScheduler： 根据TaskSet创建TaskSetManager对象存入SchedulableBuilder的数据池（Pool）中，并调用DriverEndpoint唤起消费（ReviveOffers）操作
DriverEndpoint：接受ReviveOffers指令后将TaskSet中的Tasks根据相关规则均匀分配给Executor
Executor：启动一个TaskRunner执行一个Task
 
二、Code转化为初始RDDs
          我们的用户代码通过调用Spark的Api（比如：SparkSession.builder.appName("Spark Pi").getOrCreate()），该Api会创建Spark的上下文（SparkContext）,当我们调用transform类方法 （如：parallelize(),map()）都会创建（或者装饰已有的） Spark数据结构（RDD）, 如果是action类操作（如：reduce()），那么将最后封装的RDD作为一次Job提交，存入待调度队列中（DAGSchedulerEventProcessLoop
）待后续异步处理。
          如果多次调用action类操作，那么封装的多个RDD作为多个Job提交。
     流程如下：
     


ExecuteEnv（执行环境 ）

这里可以是通过spark-submit提交的MainClass，也可以是spark-shell脚本
MainClass : 代码中必定会创建或者获取一个SparkContext
spark-shell：默认会创建一个SparkContext

RDD（弹性分布式数据集）

create：可以直接创建（如：sc.parallelize(1 until n, slices) ）,也可以在其他地方读取（如：sc.textFile("README.md")）等
transformation：rdd提供了一组api可以进行对已有RDD进行反复封装成为新的RDD，这里采用的是装饰者设计模式，下面为部分装饰器类图



action:当调用RDD的action类操作方法时（collect、reduce、lookup、save ），这触发DAGScheduler的Job提交

DAGScheduler：创建一个名为JobSubmitted的消息至DAGSchedulerEventProcessLoop阻塞消息队列（LinkedBlockingDeque）中
DAGSchedulerEventProcessLoop：启动名为【dag-scheduler-event-loop】的线程实时消费消息队列
【dag-scheduler-event-loop】处理完成后回调JobWaiter
DAGScheduler：打印Job执行结果
JobSubmitted：相关代码如下（其中jobId为DAGScheduler全局递增Id）：

eventProcessLoop.post(JobSubmitted(
jobId, rdd, func2, partitions.toArray, callSite, waiter,
SerializationUtils.clone(properties)))

最终示例：


 

     最终转化的RDD分为四层，每层都依赖于上层RDD，将ShffleRDD封装为一个Job存入DAGSchedulerEventProcessLoop待处理，如果我们的代码中存在几段上面示例代码，那么就会创建对应对的几个ShffleRDD分别存入DAGSchedulerEventProcessLoop
 
三、RDD分解为待执行任务集合（TaskSet）
     Job提交后，DAGScheduler根据RDD层次关系解析为对应的Stages，同时维护Job与Stage的关系。
     将最上层的Stage根据并发关系（findMissingPartitions ）分解为多个Task，将这个多个Task封装为TaskSet提交给TaskScheduler。非最上层的Stage的存入处理的列表中（waitingStages += stage）
 流程如下：

　　


DAGSchedulerEventProcessLoop中，线程【dag-scheduler-event-loop】处理到JobSubmitted
调用DAGScheduler进行handleJobSubmitted

首先根据RDD依赖关系依次创建Stage族,Stage分为ShuffleMapStage，ResultStage两类



更新jobId与StageId关系Map
创建ActiveJob，调用LiveListenerBug，发送SparkListenerJobStart指令
找到最上层Stage进行提交，下层Stage存入waitingStage中待后续处理

调用OutputCommitCoordinator进行stageStart()处理
调用LiveListenerBug， 发送 SparkListenerStageSubmitted指令
调用SparkContext的broadcast方法获取Broadcast对象
根据Stage类型创建对应多个Task，一个Stage根据findMissingPartitions分为多个对应的Task，Task分为ShuffleMapTask，ResultTask



将Task封装为TaskSet，调用TaskScheduler.submitTasks(taskSet)进行Task调度，关键代码如下：

taskScheduler.submitTasks(new TaskSet(
tasks.toArray, stage.id, stage.latestInfo.attemptId, jobId, properties))

 
四、TaskSet封装为TaskSetManager并提交至Driver
     TaskScheduler将TaskSet封装为TaskSetManager(new TaskSetManager(this, taskSet, maxTaskFailures, blacklistTrackerOpt)),存入待处理任务池（Pool）中，发送DriverEndpoint唤起消费（ReviveOffers）指令
      


DAGSheduler将TaskSet提交给TaskScheduler的实现类，这里是TaskChedulerImpl
TaskSchedulerImpl创建一个TaskSetManager管理TaskSet，关键代码如下：

new TaskSetManager(this, taskSet, maxTaskFailures, blacklistTrackerOpt)

同时将TaskSetManager添加SchedduableBuilder的任务池Poll中
调用SchedulerBackend的实现类进行reviveOffers，这里是standlone模式的实现类StandaloneSchedulerBackend
SchedulerBackend发送ReviveOffers指令至DriverEndpoint
 
五、Driver将TaskSetManager分解为TaskDescriptions并发布任务到Executor
     Driver接受唤起消费指令后，将所有待处理的TaskSetManager与Driver中注册的Executor资源进行匹配，最终一个TaskSetManager得到多个TaskDescription对象，按照TaskDescription想对应的Executor发送LaunchTask指令



当Driver获取到ReviveOffers（请求消费）指令时
首先根据executorDataMap缓存信息得到可用的Executor资源信息（WorkerOffer），关键代码如下

val activeExecutors = executorDataMap.filterKeys(executorIsAlive)
val workOffers = activeExecutors.map { case (id, executorData) =>
new WorkerOffer(id, executorData.executorHost, executorData.freeCores)
}.toIndexedSeq

接着调用TaskScheduler进行资源匹配，方法定义如下：

def resourceOffers(offers: IndexedSeq[WorkerOffer]): Seq[Seq[TaskDescription]] = synchronized {..}

将WorkerOffer资源打乱（val shuffledOffers = Random.shuffle(offers)）
将Poo中待处理的TaskSetManager取出（val sortedTaskSets = rootPool.getSortedTaskSetQueue），
并循环处理sortedTaskSets并与shuffledOffers循环匹配，如果shuffledOffers(i)有足够的Cpu资源（ if (availableCpus(i) >= CPUS_PER_TASK) ），调用TaskSetManager创建TaskDescription对象（taskSet.resourceOffer(execId, host, maxLocality)），最终创建了多个TaskDescription，TaskDescription定义如下：

new TaskDescription(
taskId,
attemptNum,
execId,
taskName,
index,
sched.sc.addedFiles,
sched.sc.addedJars,
task.localProperties,
serializedTask)

如果TaskDescriptions不为空，循环TaskDescriptions，序列化TaskDescription对象，并向ExecutorEndpoint发送LaunchTask指令，关键代码如下：

for (task <- taskDescriptions.flatten) {
val serializedTask = TaskDescription.encode(task)
val executorData = executorDataMap(task.executorId)
executorData.freeCores -= scheduler.CPUS_PER_TASK
executorData.executorEndpoint.send(LaunchTask(new SerializableBuffer(serializedTask)))
}