《深刻理解Spark-核心思想与源码分析》（五）第五章任务提交与执行

即欲捭之贵周，即欲阖之贵密。周密之贵，微而与道相随。---《鬼谷子》

解释：译文：若是要分析问题，关键在于周详，若是要综合概括问题，关键在于严密。周详严密的关键在于精深而与道相随。

解词：捭阖（bǎihé）：开合。意为运用某些计策和手段，使双方联合或分化。含此义的成语有“纵横捭阖”。

　　解析：《鬼谷子》是一部充满着谋略和智慧的名著。“即欲捭之贵周，即欲阖之贵密”十分鲜明地体现了此书的特色。文中首先提出了“捭”和“阖”，这是两种不一样的克敌制胜的计策。而后，再提出运用这两种计策必须具备的智慧，即“捭”要贵于周详，“阖”要贵于严密。在此基础上，又进一步指出，“周密之贵”在于“微而与道相随”。“微”意在“周密”之上，再进一步精深严密。最后提升到“道”上，则是理性的升华。

　　读此名句，在于有助于谋略的深化和智慧的提高。

5.1任务概述

1.完成RDD的转换及DAG的构建

2.完成finalStage的建立与Stage的划分，作好Stage与Task的准备工做后，最后提交Stage与Task。

3.使用集群管理器分配资源与任务调度,，对于失败的有重试和容错机制。

4.执行任务

5.2 广播Hadoop的配置信息

SparkContext的BroadCast方法用于广播Hadoop的配置信息。

  def broadcast[T: ClassTag](value: T): Broadcast[T] = {
    assertNotStopped()
    require(!classOf[RDD[_]].isAssignableFrom(classTag[T].runtimeClass),
      "Can not directly broadcast RDDs; instead, call collect() and broadcast the result.")
    val bc = env.broadcastManager.newBroadcast[T](value, isLocal)
    val callSite = getCallSite
    logInfo("Created broadcast " + bc.id + " from " + callSite.shortForm)
    cleaner.foreach(_.registerBroadcastForCleanup(bc))
    bc
  }

　　上面的代码经过使用BroadcastManager发送广播，广播结束将广播对象注册到ContextCleaner中，以便清理。

  代码中BroadcastManager的newBroadcast方法实际上代理了broadcastFactory的newBroadcast。

5.3 RDD转换及DAG构建

为何须要RDD？

下面从四个方面解释：

1.数据模型方面

RDD是一个容错的、并行的数据结构，能够控制将数据存储到磁盘或者内存，可以获取数据的分区。

2.依赖划分原则

依赖主要分为宽依赖和窄依赖，窄依赖划分为用一个Stage，管道方式迭代执行。

宽依赖的上游RDD不止一个，每每须要跨节传输数据。

3.数据处理效率

4.容错处理

RDD实现分析

5.4 任务提交

任务提交准备

1.通过5.3节对RDD的层层转换以及DAG的构建。

接下来调用RDD的collect方法转成Seq，封装为Seq为ArrayList

RDD的collect方法调用SparkContext的runJob

 

SparkContext的runJob从新调用runJob，点击runJob进入源代码

接着调用重载的runJob，最终调用的runJob方法又一次调用clean方法防止闭包的反序列化错误，而后运行dagScheduler的runJob

1.提交Job

submitJob方法将一个Job提交到Job scheduler，处理过程：
1）、调用RDD的paritition函数来获取当前Job的最大分区数，即为maxPartitions。根据maxPartitions，确认咱们没有一个不存在的partition上运行的任务

2）、生成当前Job的JobId

3）、建立JobWaiter

4）、向eventProcessActor发送JobSubmitted

5）、返回JobWaiter

2.处理Job提交

DAGSchedulerEventProcessActor收到JobSubmit事件，会调用dagScheduler的handleJobSubmitted方法。

5.4.2 finalStage的建立与Stage的划分

在Spark中，一个Job可能被划分为一个或者多个Stage，各个之间存在依赖关系，其中最下游的Stage也被成为最终的Stage，用于处理Job最后阶段的工做

1.newStage的实现分析

handleJobSubmitted方法使用newStage方法建立finalStage

2.获取父Stage列表

Spark的Job会被划分到一到多个Stage，这些Stage的划分是从finalStage开始，从后往前边划分边建立的。getParentStages方法用于获取或者建立给定的RDD的全部父Stage

这些Stage将被分配给jobId对应的job

3.获取map任务对应Stage

getShuffleMapStage方法用于建立或者获取Stage并注册到shuffleToMapStage

5.4.3 建立Job

5.4.4 提交Stage

5.4.5 提交Task

5.5 执行任务

5.5.1 状态更新

调用execBackend的statusUpdate方法更新任务状态。

5.5.2 任务还原

所谓任务还原就是将Driver提交的Task在Executor上经过反序列化、更新依赖达到Task的还原效果的过程。

5.5.3任务运行

TaskRunner最终调用Task的run方法运行任务。

5.6 任务执行后续处理

5.6.1 计量统计与执行结果序列化

5.6.2 内存回收

TaskRunner的run方法最后还会在finally中作一些清理工做。

5.6.3 执行结果处理

5.7 小结

　　首先从Spark为何设计RDD入手，依次讲解RDD的实现分析、Stage的划分、提交Stage、任务执行、执行结果处理等内容。

　　在资源分配中涉及的本地化实现，本章作了较为详细的分析，Spark经过一种阶梯式的本地化策略，

在有效利用资源、节省网络I/O的同时提升系统执行的效率。

　　容错方面，Spark经过DAG构成的有向无环图能够在某些任务执行失败的状况下，经过从新提交任务达到容错，而那些执行成功的任务因为结果已经存在缓存中，因此不须要重复计算。