Spark Core内核调度机制详解(第5天）

系列文章目录

  1. 如何构建DAG执行流程图 (掌握)
  2. 如何划分Stage阶段 (掌握)
  3. Driver底层是如何运转 (掌握)
  4. 确定需要构建多少分区(线程) (掌握)

文章目录

  • 系列文章目录
  • 引言
  • 一、Spark内核调度（掌握）
      • 1.1、内容概述
      • 1.2、RDD的依赖
      • 1.3、DAG和Stage
      • 1.4、Spark Shuffle
      • 1.5、Job调度流程
      • 1.6、Spark RDD并行度
  • 二. 常见面试题
      • 1.简单介绍下RDD宽依赖和窄依赖的区别
      • 2.简单介绍下DAG有向无环图如何划分Stage阶段的
      • 3.请描述下rdd中job的整个调度流程

引言

本文主要介绍了
1.RDD的依赖
2.DAG和Stage
3.Spark Shuffle
4.job调度流程(掌握)
5.Spark RDD并行度
帮助读者更好地理解其工作原理和优化方法。

一、Spark内核调度（掌握）

1.1、内容概述

Spark内核调度的任务：

  • 如何构建DAG执行流程图
  • 如何划分Stage阶段
  • Driver底层是如何运转
  • 确定需要构建多少分区（线程）

Spark内核调度的目的：尽可能用最少的资源高效地完成任务计算。

1.2、RDD的依赖

RDD依赖：一个RDD的形成可能是由一个或者多个RDD得到的，此时这个RDD和之前的RDD之间产生依赖关系。

在Spark中，RDD之间的依赖关系，主要有二种类型：

  • 窄依赖
  1. 作用: 能够让Spark程序并行计算。也就是一个分区数据计算出现问题以后，其他的分区计算不受到任何影响。
  2. 特点: 父RDD的分区和子RDD的分区关系是一对一的关系。也就是父RDD分区的数据会整个被下游子RDD的分区接收。

在这里插入图片描述

  • 宽依赖
  1. 作用: 划分Stage的重要依据。宽依赖也叫做Shuffle依赖。
  2. 特点: 父RDD的分区和子RDD的分区关系是一对多的关系，也就是父RDD的分区数据会被分成多份给到下游子RDD的多个分区所接收。
  3. 注意:为了避免数据不完整，如果有宽依赖，shuffle下游的其他操作，必须等待shuffle执行完成以后才能够继续执行。

在这里插入图片描述

  • 说明
    1. 在实际使用中，不需要纠结哪些算子会存在shuffle，以需求为目标。虽然shuffle的存在会影响一定的效率, 但是以完成任务为准则，该用那个算子，就使用那个算子即可，不要过分纠结。
    2. 算子中一般以ByKey结尾的会发生shuffle另外是重分区算子也会发生shuffle。

1.3、DAG和Stage

  • DAG：有向无环图，主要描述一段执行任务，从开始一直往下走，不允许出现回调操作,Spark应用程序中，遇到一个Action算子，就会触发形成一个Job任务的产生。

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

层级关系：
1- 一个application应用程序 -> 遇到一个Action算子，就会触发形成一个Job任务
2- 一个Job任务只有一个DAG有向无环图
3- 一个DAG有向无环图 -> 有多个Stage
4- 一个Stage -> 有多个Task线程5- 一个RDD -> 有多个分区
6- 一个分区会被一个Task线程所处理
  • DAG执行流程图形成和Stage划分
    在这里插入图片描述
1- Spark应用程序遇到Action算子后，就会触发一个Job任务的产生。Job任务会将它所依赖的所有算子全部加载进来，形成一个Stage。2- 接着从Action算子从后往前进行回溯，遇到窄依赖就将算子放在同一个Stage当中；如果遇到宽依赖，就划分形成新的Stage，最后一直回溯完成。
  • 细化剖析Stage内部的流程
    在这里插入图片描述
  • 默认并行度的值确认
因为是使用textFile读取HDFS上的文件，因此RDD分区数=max(文件的block块的数量, defaultMinPartition)，继续需要知道defaultMinPartition的值是多少。defaultMinPartition=min(spark.default.parallelism,2)取最小值。最终我们确认spark.default.parallelism的参数值就能够最终确认RDD的分区数有多少个。spark.default.parallelism参数值确认过程如下：
1- 如果有父RDD，就取父RDD的最大分区数
2- 如果没有父RDD，根据集群模式进行取值：2.1- 本地模式：机器的最大CPU核数2.2- （了解）Mesos：默认是82.3- 其他模式：所有执行节点上的核总数或2，以较大者为准

1.4、Spark Shuffle

  • Spark中shuffle的发展历程
1- 在1.1版本以前，Spark采用Hash shuffle (优化前 和 优化后)2- 在1.1版本的时候，Spark推出了Sort Shuffle3- 在1.5版本的时候，Spark引入钨丝计划(优化为主)4- 在1.6版本的时候，将钨丝计划合并到sortShuffle中5- 在2.0版本的时候，将Hash Shuffle移除，将Hash shuffle方案移植到Sort Shuffle

在这里插入图片描述

  • 未优化的Hash shuffle
    在这里插入图片描述

  • 存在的问题
    上游（map端）的每个Task会产生与下游Task个数相等的小文件个数。这种情况会导致上游有非常多的小文件。另外，下游（reduce端）来拉取文件的时候，会有大量的网络IO和磁盘IO过程，因为要打开和读取多个小文件。

  • 经过优化后的Hash shuffle
    在这里插入图片描述
    变成了由每个Executor进程产生与下游Task个数相等的小文件数。这样可以大量减小小文件的产生，以及降低下游拉取文件时候的网络IO和磁盘IO过程。

  • Sort shuffle
    在这里插入图片描述

Sort Shuffle分成了两种: 普通机制和bypass机制。具体使用哪种，由Spark底层决定。普通机制的运行过程: 每个上游Task线程处理数据，数据处理完以后，先放在内存中。接着对内存中的数据进行分区、排序。将内存中的数据溢写到磁盘，形成一个个的小文件。溢写完成以后，会将多个小文件合并成一个大的磁盘文件。并且针对每个大的磁盘文件，会提供一个索引文件。接着是下游Task根据索引文件来读取相应的数据。bypass机制: 就是在普通机制的基础上，省略了排序的过程bypass机制的触发条件是：
1- 上游RDD的分区数量最多不能超过200个
2- 上游不能对数据进行提前聚合操作（因为提前聚合，需要先进行分组操作，而分组的操作实际上是有排序的操作）

1.5、Job调度流程

  • 主要是讨论：在Driver内部，是如何调度任务
    在这里插入图片描述
1- Driver进程启动后，底层PY4J创建SparkContext顶级对象。在创建该对象的过程中，还会创建另外两个对象，分别是: DAGScheduler和TaskSchedulerDAGScheduler: DAG调度器。将Job任务形成DAG有向无环图和划分Stage的阶段TaskScheduler: Task调度器。将Task线程分配给到具体的Executor执行2- 一个Spark程序遇到一个Action算子就会触发产生一个Job任务。SparkContext将Job任务给到DAG调度器，拿到Job任务后，会将Job任务形成DAG有向无环图和划分Stage的阶段。并且会确定每个Stage阶段有多少个Task线程，会将众多的Task线程放到TaskSet的集合中。DAG调度器将TaskSet集合给到Task调度器3- Task调度器拿到TaskSet集合以后，将Task分配给到给到具体的Executor执行。底层是基于SchedulerBackend调度队列来实现的。4- Executor开始执行任务。并且Driver会监控各个Executor的执行状态，直到所有的Executor执行完成，就认为任务运行结束5- 后续过程和之前一样

1.6、Spark RDD并行度

  • 整个Spark应用中，影响并行度的因素有以下两个原因:
    1- 资源的并行度: Executor数量 和 CPU核心数 以及 内存的大小。
    2- 数据的并行度: Task的线程数 和 分区数量。
一般将Task线程数设置为CPU核数的2-3倍。另外每个线程分配3-5GB的内存资源。
  • 如何设置并行度
语法: SparkConf().set("spark.default.parallelism", "num")说明: spark.default.parallelism该参数是SparkCore中的参数。
该参数只会影响shuffle以后的分区数量。
另外该参数对parallelize并行化本地集合创建的RDD不起作用。
  • 代码演示
# 导包
from pyspark import SparkContext, SparkConf
import os# 绑定指定的python解释器
os.environ['SPARK_HOME'] = '/export/server/spark'
os.environ['PYSPARK_PYTHON'] = '/root/anaconda3/bin/python3'
os.environ['PYSPARK_DRIVER_PYTHON'] = '/root/anaconda3/bin/python3'
if __name__ == '__main__':# 6.1 TODO: 设置并行度conf = SparkConf().set('spark.default.parallelism', '6')# - 1.创建SparkContext对象sc = SparkContext(conf=conf)# - 2.数据输入textRDD = sc.textFile('file:///export/data/spark_project/spark_base/content.txt')# - 3.数据处理#   - 3.1文本内容切分flatMapRDD = textRDD.flatMap(lambda line: line.split(" "))#   - 3.2数据格式转换mapRDD = flatMapRDD.map(lambda word: (word, 1))# 6.2 TODO: shuffle之前查看分区数量print(f"shullfe之前分区数: {mapRDD.getNumPartitions()},分区内容:{mapRDD.glom().collect()}")#   - 3.3分组和聚合reduceRDD = mapRDD.reduceByKey(lambda agg, curr: agg + curr)# 6.3 TODO: shuffle之后查看分区数量print(f"shullfe之后分区数: {reduceRDD.getNumPartitions()},分区内容:{reduceRDD.glom().collect()}")# - 4.数据输出# print(reduceRDD.collect())# - 5.释放资源sc.stop()

二. 常见面试题

1.简单介绍下RDD宽依赖和窄依赖的区别

--- 窄依赖（Narrow Dependency）
1- 定义：每一个父RDD的分区最多被子RDD的一个分区所使用。这通常表现为一对一（OneToOneDependencies）或多对一（多个父RDD的分区对应于一个子RDD的分区）的关系。
2- 操作示例：map、filter、union等操作会产生窄依赖。这些操作的特点是，子RDD的每个分区都是基于父RDD的一个或多个特定分区计算得出的。
3- 容错处理：当窄依赖的子RDD数据丢失时，由于父RDD的一个分区只对应一个子RDD分区，因此只需要重新计算与子RDD分区对应的父RDD分区即可。
4- 优化特性：窄依赖可以进行流水线优化，利用fork/join机制，一个作业可以直接一个阶段完成，形成管道型的流水化处理。
--- 宽依赖（Wide Dependency 或 Shuffle Dependency）
1- 定义：子RDD的每一个分区都会使用所有父RDD的所有分区或多个分区。这通常表现为一对多（OneToManyDependencies）的关系。
2- 操作示例：groupByKey、reduceByKey、sortedByKey等操作会产生宽依赖。这些操作的特点是，子RDD的每个分区都可能依赖于父RDD的所有分区。
3- 容错处理：当宽依赖的子RDD数据丢失时，由于一个分区的数据通常由多个父RDD分区数据变换而来，因此可能需要重新计算父RDD的所有分区才能恢复数据。
4- 优化限制：宽依赖通常涉及shuffle操作，需要将数据写入磁盘并等待，因此无法形成管道型的流水化处理。
  • 总结
    数据流动：窄依赖的数据流动是线性的，而宽依赖的数据流动是跨多个分区的。
    容错处理：窄依赖的容错处理更为高效，只需要重新计算部分数据；而宽依赖的容错处理可能涉及大量数据的重新计算。
    优化特性：窄依赖可以进行流水线优化，而宽依赖通常不能。
    Shuffle操作：宽依赖通常伴随着shuffle操作，而窄依赖则没有。
    以上是对RDD宽依赖和窄依赖的简要介绍和区别分析。

2.简单介绍下DAG有向无环图如何划分Stage阶段的

  • 1- Spark应用程序遇到Action算子后，就会触发一个Job任务的产生。Job任务会将它所依赖的所有算子全部加载进来，形成一个Stage。

  • 2- 接着从Action算子从后往前进行回溯，遇到窄依赖就将算子放在同一个Stage当中；如果遇到宽依赖，就划分形成新的Stage，最后一直回溯完成。

3.请描述下rdd中job的整个调度流程

1- Driver进程启动后，底层PY4J创建SparkContext顶级对象。在创建该对象的过程中，还会创建另外两个对象，分别是: DAGScheduler和TaskSchedulerDAGScheduler: DAG调度器。将Job任务形成DAG有向无环图和划分Stage的阶段TaskScheduler: Task调度器。将Task线程分配给到具体的Executor执行2- 一个Spark程序遇到一个Action算子就会触发产生一个Job任务。SparkContext将Job任务给到DAG调度器，拿到Job任务后，会将Job任务形成DAG有向无环图和划分Stage的阶段。并且会确定每个Stage阶段有多少个Task线程，会将众多的Task线程放到TaskSet的集合中。DAG调度器将TaskSet集合给到Task调度器3- Task调度器拿到TaskSet集合以后，将Task分配给到给到具体的Executor执行。底层是基于SchedulerBackend调度队列来实现的。4- Executor开始执行任务。并且Driver会监控各个Executor的执行状态，直到所有的Executor执行完成，就认为任务运行结束5- 后续过程和之前一样

本文来自互联网用户投稿，该文观点仅代表作者本人，不代表本站立场。本站仅提供信息存储空间服务，不拥有所有权，不承担相关法律责任。如若转载，请注明出处：http://www.rhkb.cn/news/357544.html

如若内容造成侵权/违法违规/事实不符，请联系长河编程网进行投诉反馈email:809451989@qq.com，一经查实，立即删除！

相关文章

云计算 私有云openstack 搭建指导 【99%成功率】

云计算 私有云openstack 搭建指导 【99%成功率】

本人已经搭了很多次，也失败了很多次，这里面有很多失败的经验，以及提示，请按提示搭建，相信你也可以的！ 不积小流无以成江海，多尝试才能吸取经验 开始搭建openstack 准备文件（如果没有…
阅读更多...
练手代码之使用Python实现合并PDF文件

练手代码之使用Python实现合并PDF文件

如果你有合并PDF的需要，你会怎么办 我们无所不能的程序员会选择写一个Python代码来实现（谁会这么无聊？是我），如果真的有PDF操作需要，我推荐你使用PDF Expert这个软件哈~ 话不多说直接上代码： …
阅读更多...
Pip换源秘籍：让你的Python包飞行起来!

Pip换源秘籍：让你的Python包飞行起来!

在Python的包管理中，Pip是最重要的工具之一。它允许开发者从Python Package Index (PyPI)安装包，但有时由于网络问题或服务器负载过高，直接从PyPI安装包可能会非常慢。这时，更换Pip源到一个更快的镜像站点是一个常见的解决方案。本…
阅读更多...
版本控制工具-git的基本使用

版本控制工具-git的基本使用

目录 前言一、git简介二、git工作流程三、git基本命令3.1 创建本地仓库3.2 将工作区内容提交到本地仓库3.3 将本地仓库内容推送到远程仓库 前言 本篇文章介绍git的一般工作流程 一、git简介 Git是一个开源的分布式版本控制软件，常用于项目的版本管理 Git是Linux …
阅读更多...
破碎的像素地牢探险：游戏分享

破碎的像素地牢探险：游戏分享

软件介绍 《破碎的像素地牢》是开源一款地牢冒险探索类的游戏，融合了日系RPG经典风格，玩家将控制主角进行未知场景的探索。除了经典地牢玩法外，游戏还添加了更多创意内容，如黑屏状态前的挑战性等，使得游戏更加富有挑战…
阅读更多...
AI智能时代：ChatGPT如何在金融市场发挥策略分析与预测能力？

AI智能时代：ChatGPT如何在金融市场发挥策略分析与预测能力？

文章目录 一、ChatGPT在金融策略制定中的深度应用客户需求分析与定制化策略市场动态跟踪与策略调整策略分析与优化 二、ChatGPT在算法交易中的深度应用自动交易策略制定交易执行与监控风险管理 三、未来展望《智能量化：ChatGPT在金融策略与算法交易中的实践》亮点内…
阅读更多...
Windows10中端口被占用处理方法

Windows10中端口被占用处理方法

前言 在Windows 10中，查看端口被占用情况的方法主要依赖于命令行工具netstat。以下是详细步骤，以及必要的解释和归纳： 打开命令提示符 方法1：使用快捷键Win R，打开“运行”对话框，输入cmd，然…
阅读更多...
Linux 进程管理

Linux 进程管理

一、查看进程 使用ps -aux进行查看，其中a表示列出所有进程信息，u以用户格式显示进程信息，x显示后台进程参数，也可以使用| grep 进行进程的筛选 以下是显示进程后的示意 USER为进程执行的用户 PID为进程号 %CPU为该进程的cpu占用…
阅读更多...
如何解决vue中的路由守卫失效问题

如何解决vue中的路由守卫失效问题

引言 1. 路由守卫简介 路由守卫是前端开发中一个至关重要的概念，特别是在使用单页应用（SPA）框架如React、Vue或Angular时。它们充当了SPA中的“门卫”，控制着用户对不同页面的访问权限。路由守卫的核心功能是确保用户在访问特定…
阅读更多...
挖矿宝藏之开发者模式

挖矿宝藏之开发者模式

目录 一、开发者模式简介 二、启动方式 三、元素（Elements） 四、控制台（Console） 五、来源（Sources） 六、网络（Network） 七、性能（Performance） 八、…
阅读更多...
Day10—Spark SQL基础

Day10—Spark SQL基础

Spark SQL介绍 ​ Spark SQL是一个用于结构化数据处理的Spark组件。所谓结构化数据，是指具有Schema信息的数据，例如JSON、Parquet、Avro、CSV格式的数据。与基础的Spark RDD API不同，Spark SQL提供了对结构化数据的查询和计算接口。 Spark …
阅读更多...
react18 实现具名插槽

react18 实现具名插槽

效果预览 技术要点 当父组件给子组件传递的 JSX 超过一个标签时，子组件接收到的 children 是一个数组，通过解析数组中各 JSX 的属性 slot ，即可实现具名插槽的分发！ 代码实现 Father.jsx import Child from "./Child";…
阅读更多...
OGG几何内核开发-复杂装配模型读取、显示、分析

OGG几何内核开发-复杂装配模型读取、显示、分析

OGG几何内核读取STEP模型文件的API有STEPCAFControl_Reader、STEPControl_Reader。 STEPCAFControl_Reader使用很复杂，但可以展示装配树，有利于模型的详细分析。 本文演示了《插件化算法研究平台V2》的OCC几何模型插件的部分功能：显示装配树…
阅读更多...
Golang | Leetcode Golang题解之第172题阶乘后的零

Golang | Leetcode Golang题解之第172题阶乘后的零

题目： 题解： func trailingZeroes(n int) (ans int) {for n > 0 {n / 5ans n}return }
阅读更多...
pytets测试框架中如果需要运行多个测试套件时pytest.ini文件设置

pytets测试框架中如果需要运行多个测试套件时pytest.ini文件设置

pytets测试框架中如果需要运行多个测试套件时pytest.ini文件设置方法 testpaths testcases/fenmi testcases/weixin testcases/Zgen
阅读更多...
QT实现人脸识别

QT实现人脸识别

QT实现人脸识别 Face.pro文件： QT core guigreaterThan(QT_MAJOR_VERSION, 4): QT widgetsCONFIG c11# The following define makes your compiler emit warnings if you use # any Qt feature that has been marked deprecated (the exact warnings # d…
阅读更多...
【MATLAB】（高数）

【MATLAB】（高数）

参考文章 函数极限 导数与偏导 极值和最值 局部范围的最值 局部范围内的最值，相当于函数的极值 离散数据的最值 多元函数的极值 fminunc [x, fval] fminunc(fun, x0)fun为代求极值的函数；x0为起始点，即从这个点开始寻找极值，…
阅读更多...
数据结构9——排序

数据结构9——排序

一、冒泡排序 冒泡排序（Bubble Sort），顾名思义，就是指越小的元素会经由交换慢慢“浮”到数列的顶端。 算法原理 从左到右，依次比较相邻的元素大小，更大的元素交换到右边；从第一组相邻元素比较…
阅读更多...
贪心算法——赶作业（C++）

贪心算法——赶作业（C++）

慢慢来，沉稳一点。 2024年6月18日 题目描述 A同学有n份作业要做，每份作业有一个最后期限，如果在最后期限后交作业就会扣分，现在假设完成每份作业都需要一天。A同学想安排作业顺序，把扣分降到最低，请帮他实…
阅读更多...
工业园安全生产新保障：广东地区加强可燃气体报警器校准检测

工业园安全生产新保障：广东地区加强可燃气体报警器校准检测

广东，作为我国经济的重要引擎，拥有众多工业园区。 这些工业园区中，涉及化工、制药、机械制造等多个领域，每天都会产生和使用大量的可燃气体。因此，可燃气体报警器的安装与校准检测，对于保障工业园区的安全…
阅读更多...
最新文章
推荐文章

Copyright @ 2022~2023