Pyspark

注：大家觉得博客好的话，别忘了点赞收藏呀，本人每周都会更新关于人工智能和大数据相关的内容，内容多为原创，Python Java Scala SQL 代码，CV NLP 推荐系统等，Spark Flink Kafka Hbase Hive Flume等等~写的都是纯干货，各种顶会的论文解读，一起进步。
今天继续和大家分享一下Pyspark基础入门7
#博学谷IT学习技术支持

---

---

前言

今天和大家分享的是Spark RDD的内核调度

---

一、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的存在, 会影响一定的效率,但是以完成需求为准则, 该用那个算子, 就使用那个算子即可, 不要过分纠结

二、DAG和STAGE

DAG: 有向无环图
主要描述一段执行任务, 从开始一直往下 执行, 不允许出现回调的操作
在Spark的应用程序中, 程序中有一个action算子, 就会触发一个Job任务,所以说一个Spark应用程序中可以有多个Job任务
对于每一个Job任务, 都会产生一个DAG执行流程图

• 第一步: 当Spark应用遇到一个action算子后, 就会触发一个Job任务执行,
  首先会将这个action算子所依赖的所有的RDD全部都加载进行, 形成一个完整的stage阶段
• 第二步: 根据RDD之间的宽窄依赖关系, 从后往前进行回溯,如果遇到窄依赖, 就放置在一起, 形成一个stage, 如果遇到宽依赖,
  就拆分为两个阶段,直到回溯完成, 形成最终的DAG执行流程图

三、RDD的shuffle

在2.0版本的时候, 将Hash shuffle剔除, 将Hash Shuffle方案被Sort Shuffle

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

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

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

bypass机制使用条件:
1- 上游的分区的数量不能超过200个(默认)
2- 上游不能进行提前聚合操作(提前聚合意味着要进行分组操作, 而分组的前提是要对数据进行排序, 将相关的数据放置在一起)

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

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

四、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执行状态即可, 等待任务执行完成

五、Spark的并行度

整个Spark应用中, 影响程序并行度的因素有以下两个原因:

• 1- 资源的并行度: Executor数量 和 CPU核心数 以及 内存大小
• 2- 数据的并行度: Task的线程数 和 分区数量

目的: 在合适的资源上, 运行合适的数据

当资源比较充足的时候, 但是数据的并行度无法达到, 虽然不会影响效率 但是会浪费资源
当数据的并行度充足的时候, 而资源不足, 会导致本来可以并行计算的, 变成串行执行, 影响效率

推荐值:
一个CPU上运行 2~3个线程, 一个CPU需要配置3~5GB内存, 从而充分压榨CPU性能
一个线程, 建议处理1~3GB数据, 处理时间10~20分钟, 如果想要时间减少, 那么让每一个线程处理数据量减少

说明: 并行度设置, 需要在shuffle后生效, shuffle前的分区数量, 默认取决于初始数据源的时候确定的分区数量(上一个父RDD的分区数量)

---

总结

今天主要和大家分享了RDD的内核调度。