Spark六：Spark 底层执行原理SparkContext、DAG、TaskScheduler

Spark底层执行原理

学习Spark运行流程
学习链接：https://mp.weixin.qq.com/s/caCk3mM5iXy0FaXCLkDwYQ

一、Spark运行流程

流程：

1. SparkContext想西苑管理器注册并向资源管理器申请运行Executor
2. 资源管理器分配Executor，然后资源管理器启动Executor
3. Executor发送心跳至资源管理器
4. SparkContext构建DAG有向无环图
5. 将DAG分解成Stage(TaskSet)
6. 把Stage发送给TaskScheduler
7. Executor向SparkContext申请Task
8. TaskScheduler将Task发送给Executor运行
9. 同时SparkContext将应用程序代码发给Executor
   10.Task在Executor上运行，运行完毕后释放所有资源

1.1 从代码角度看DAG图的构建

val session = SparkSession.builder().master("local[*]").appName().getOrCreate()
val sc = session.sparkContext
val lines1 = sc.textFile(inputPath1).map(...).map(...)
val lines2 = sc.textFile(inputPath2).map(...)
val lines3 = sc.textFile(inputPath3)

val dtinone1 = lines2.union(lines3)
val dtinone = lines1.join(dtinone1)
dtinone.saveAsTextFile(...)
dtinone.filter(...).foreach(...)

代码的DAG图：

Spark内核会在需要计算发生的时刻绘制一张关于计算路径的有向无环图，也就是上图所示的DAG。

Spark的计算发生在RDD的Action操作，而对Action之前所有的Transformation，Spark只是记录下RDD生成的轨迹，不会触发真正的计算。

1.2 将DAG划分为Stage核心算法

一个Application可以有多个job多个Stage：
Spark Application中可以因为有不同的Action触发众多的job，一个Application中可以有很多的job，每个job是由一个或者多个Stage构成的，后面的Stage依赖于前面的Stage，也就是只有前面的Stage计算完毕后，后面的Stage才会运行。

---

划分依据：
Stage划分的一句是宽依赖：像像 reduceByKey，groupByKey 等算子，会导致宽依赖的产生。

宽窄依赖划分原则：
窄依赖：父 RDD 的一个分区只会被子 RDD 的一个分区依赖。即一对一或者多对一的关系，可理解为独生子女。常见的窄依赖有：map、filter、union、mapPartitions、mapValues、join（父 RDD 是 hash-partitioned）等。
宽依赖：父 RDD 的一个分区会被子 RDD 的多个分区依赖(涉及到 shuffle)。即一对多的关系，可理解为超生。常见的宽依赖有 groupByKey、partitionBy、reduceByKey、join（父 RDD 不是 hash-partitioned）等。

---

核心算法：回溯算法
从后往前回溯/反向解析，遇到窄依赖加入本Stage，遇到宽依赖进行Stage切分。
Spark内核会从触发Action操作的那个RDD开始，从后往前推，首先回味最后一个RDD创建要一个Stage，然后倒推，如果发现对某个RDD是宽依赖，那么会将宽依赖的那个RDD创建一个新的Stage，那个RDD就是新的Stage的最后一个RDD，以此类推继续倒推，知道所有的RDD全部遍历完成。

1.3 DAG划分为Stage剖析

具体可见 Spark二、Spark技术栈之Spark Core 第六点。

1.4 提交Stages

调度阶段的提交，最终会被转换成一个任务的提交：

• DAGScheduler通过TaskScheduler接口提交任务集
• 这个任务集最终会触发TaskScheduler构建一个TaskSetManager的实例来管理这个任务集的生命周期
• 对于DAGScheduler来说，提交调度阶段的工作到此就完成了。

而TaskScheduler的具体实现则会在得到计算资源的时候，进一步通过TaskSetManager调度具体的任务到对应的Executor节点上进行运算。

1.5 监控Job、Task、Executor

1.5.1 DAGScheduler监控Job与Task

要保证相互依赖的作业调度阶段能够得到顺利的调度执行，DAGScheduler 需要监控当前作业调度阶段乃至任务的完成情况。

---

通过对外暴露一系列的回调函数实现，对于TaskScheduler来说，这些回调任务主要包括任务的开始结束失败、任务集的失败，DAGScheduler 根据这些任务的生命周期信息进一步维护作业和调度阶段的状态信息。

1.5.2 DAGScheduler 监控 Executor 的生命状态

TaskScheduler 通过回调函数通知 DAGScheduler 具体的 Executor 的生命状态，如果某一个 Executor 崩溃了，则对应的调度阶段任务集的 ShuffleMapTask 的输出结果也将标志为不可用，这将导致对应任务集状态的变更，进而重新执行相关计算任务，以获取丢失的相关数据。

1.6 获取任务执行结果

1.6.1 结果DAGScheduler

一个具体的任务在 Executor 中执行完毕后，其结果需要以某种形式返回给 DAGScheduler，根据任务类型的不同，任务结果的返回方式也不同。

1.6.2 两种结果：中间结果与最终结果

1. 对于 FinalStage 所对应的任务，返回给 DAGScheduler 的是运算结果本身。
2. 对于中间调度阶段对应的任务 ShuffleMapTask，返回给 DAGScheduler 的是一个 MapStatus 里的相关存储信息，而非结果本身，这些存储位置信息将作为下一个调度阶段的任务获取输入数据的依据。

1.6.3 两种类型：DirectTaskResult 与 IndirectTaskResult

按任务结果大小的不同，ResultTask返回的结果分成两类：

1. 如果结果足够小，则直接放在 DirectTaskResult 对象内中。
2. 如果超过特定尺寸则在 Executor 端会将 DirectTaskResult 先序列化，再把序列化的结果作为一个数据块存放在 BlockManager 中，然后将 BlockManager 返回的 BlockID 放在 IndirectTaskResult 对象中返回给 TaskScheduler，TaskScheduler 进而调用 TaskResultGetter 将 IndirectTaskResult 中的 BlockID 取出并通过 BlockManager 最终取得对应的 DirectTaskResult。

1.7 任务调度总体诠释

二、Spark运行架构特点

2.1 Executor进程专属

每个Application获取专属的Executor进程，该进程在Application期间一直驻留，并以多线程方式运行Tasks。

Spark Application不能跨应用程序共享数据，除非将数据写入到外部存储系统，如图所示：

2.2 支持多种资源管理器

Spark与资源管理器无关，只要能够获取Executor进程，并能保持互相通信就可以了。
Spark支持支援管理器包括：Standalone、On Mesos、On YARN、Or On EC2。如图所示：

2.3 Job提交就近原则

提交SparkContext的Client应该靠近Worker节点（运行Executor的节点），最好在同一个Rack（机架）里，因为Spark Application运行过程中SparkContext和Executor之间有大量的信息交换；

如果想在远程集群中运行，最好使用RPC将SparkContext提交给集群，不要远离Worker运行SparkContext。

2.4 移动程序而非移动数据的原则执行

移动程序而非移动数据的原则知性，Task采用了数据本地行和推测知性的优化机制。
关键方法：taskIdToLocations、getPreferedLocations。