RDD的内核调度【博学谷学习记录】

1. RDD的依赖关系

RDD的依赖: 指的一个RDD的形成可能是有一个或者多个RDD得出, 此时这个RDD和之前的RDD之间产生依赖关系

在Spark中, RDD之间的依赖关系,主要有二种依赖关系:

1- 窄依赖:

目的: 为了实现并行计算操作, 并且提高容错的能力

指的: 一个RDD上的一个分区的数据, 只能完整的交付给下一个RDD的一个分区(完全继承),不能分隔

2- 宽依赖:

目的: 划分stage的依据

指的: 上一个RDD的某一个分区的数据被下游的一个RDD的多个分区的所接收, 中间必然存在shuffle操作(是否存在shuffle操作是判定宽窄依赖关系的重要依据)

注意: 一旦有了shuffle操作, 后续的RDD的执行必须等待前序的RDD(shuffle)执行完成,才能执行

在Spark中, 每一个算子是否会执行shuffle操作, 其实Spark在设计算子的时候, 就已经规划好了, 比如说: Map算子就不会触发shuffle, reduceByKey算子一定会触发shuffle操作

如果想知道这个算子是否会触发shuffle操作, 可以通过在运行的时候, 查看默认4040 WEB UI界面. 在界面中对应Job的DAG执行流程图中, 如果这个图被划为为了多个stage, 那么就说明这个算子会触发shuffle. 或者也可以查看这个算子源码. 一般在源码的说明信息中也会有一定的标记是否有shuffle

在实际使用中, 不需要纠结哪些算子会存在shuffle, 以需求实现为目标, 虽然shuffle的存在, 会影响一定的效率,但是以完成需求为准则, 该用那个算子, 就使用那个算子即可, 不要过分纠结

2. DAG和STAGE

DAG: 有向无环图

主要描述一段执行任务, 从开始一直往下 执行, 不允许出现回调的操作

在Spark的应用程序中, 程序中有一个action算子, 就会触发一个Job任务,所以说一个Spark应用程序中可以有多个Job任务

对于每一个Job任务, 都会产生一个DAG执行流程图, 那么这个执行流程图是如何形成的呢?

第一步: 当Spark应用遇到一个action算子后, 就会触发一个Job任务执行, 首先会将这个action算子所依赖的所有的RDD全部都加载进行, 形成一个完整的stage阶段

第二步: 根据RDD之间的宽窄依赖关系, 从后往前进行回溯,如果遇到窄依赖, 就放置在一起, 形成一个stage, 如果遇到宽依赖, 就拆分为两个阶段,直到回溯完成, 形成最终的DAG执行流程图

3. RDD的shuffle

Sort Shuffle执行流程 与 MR有非常高的相似度:

每个线程(分区)处理后, 将数据写入到内存中, 当内存数据达到一定的阈值后, 触发溢写操作,在一些的时候, 需要对数据进行分区/排序, 将数据写入到磁盘上, 形成一个个文件, 当整个溢写完成后, 将多个小文件合并为一个大文件, 同时会为这个大文件提供一个索引文件, 方便下游读取对应分区的数据

Sort shuffle 存在两种运行的机制: 普通机制 和 byPass机制

普通机制:

每个线程(分区)处理后, 将数据写入到内存中, 当内存数据达到一定的阈值后, 触发溢写操作,在一些的时候, 需要对数据进行分区/排序, 将数据写入到磁盘上, 形成一个个文件, 当整个溢写完成后, 将多个小文件合并为一个大文件, 同时会为这个大文件提供一个索引文件, 方便下游读取对应分区的数据

bypass机制使用条件:

1- 上游的分区的数量不能超过200个(默认)

2- 上游不能进行提前聚合操作(提前聚合意味着要进行分组操作, 而分组的前提是要对数据进行排序, 将相关的数据放置在一起)

bypass机制: 在普通的机制基础上, 去除了排序操作

两种机制, bypass的运行效率在某些条件下, 可能要优于普通机制

4. Job的调度流程

1- 当Spark应用程序启动后, 此时首先会创建SparkContext对象, 此对象在创建的时候, 底层同时也会创建DAGScheduler 和 TaskScheduler:

DAGScheduler: 负责DAG流程图生成, Stage阶段划分, 每个阶段运行多少个线程

TaskScheduler: 负责每个阶段的Task线程的分配工作, 以及将对应线程任务提交到Executor上运行

2- 遇到Action算子后,就会产生一个Job任务, SparkContext对象将任务提交到DAGScheduler,DAGScheduler接收到任务后, 就会产生一个DAG执行流程图, 划分stage,并且确定每个stage中需要运行多少个线程,将每个阶段的线程放置到一个TaskSet集合中,提交给TaskScheduler

3- TaskScheduler接收到各个阶段的TaskSet后, 开始进行任务的分配工作,确认每个线程应该运行在那个executor上(尽可能保持均衡),然后将任务提交给对应executor上(底层由调度队列), 让executor启动线程执行任务即可,阶段是一个一个的运行, 无法并行执行的

4- Driver负责监听各个executor执行状态即可, 等待任务执行完成