广东建设工程备案网站,企业网站建设多长时间,徐州建设公司网站,设计公司企业想法描述Pyspark
注#xff1a;大家觉得博客好的话#xff0c;别忘了点赞收藏呀#xff0c;本人每周都会更新关于人工智能和大数据相关的内容#xff0c;内容多为原创#xff0c;Python Java Scala SQL 代码#xff0c;CV NLP 推荐系统等#xff0c;Spark Flink Kafka Hbase Hi…Pyspark
注大家觉得博客好的话别忘了点赞收藏呀本人每周都会更新关于人工智能和大数据相关的内容内容多为原创Python Java Scala SQL 代码CV NLP 推荐系统等Spark Flink Kafka Hbase Hive Flume等等~写的都是纯干货各种顶会的论文解读一起进步。 今天继续和大家分享一下Pyspark基础入门7 #博学谷IT学习技术支持 文章目录Pyspark前言一、RDD的依赖关系1 窄依赖2 宽依赖二、DAG和STAGE三、RDD的shuffle四、Job的调度流程五、Spark的并行度总结前言
今天和大家分享的是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的内核调度。
