Spark任务调度机制论述(转载)

2024-06-17 12:38
文章标签 机制 转载 任务调度 spark 论述

本文主要是介绍Spark任务调度机制论述(转载)，希望对大家解决编程问题提供一定的参考价值，需要的开发者们随着小编来一起学习吧！

Spark任务调度机制论述

在生产环境下，Spark集群的部署方式一般为YARN-Cluster模式。 Driver线程主要是初始化SparkContext对象，准备运行所需的上下文，然后一方面保持与ApplicationMaster的RPC连接，通过ApplicationMaster申请资源，另一方面根据用户业务逻辑开始调度任务，将任务下发到已有的空闲Executor上。
当ResourceManager向ApplicationMaster返回Container资源时，ApplicationMaster就尝试在对应的Container上启动Executor进程，Executor进程起来后，会向Driver反向注册，注册成功后保持与Driver的心跳，同时等待Driver分发任务，当分发的任务执行完毕后，将任务状态上报给Driver。

1. Spark任务调度概述

1.1 基础概念

当Driver起来后，Driver则会根据用户程序逻辑准备任务，并根据Executor资源情况逐步分发任务。在详细阐述任务调度前，首先说明下Spark里的几个概念。一个Spark应用程序包括Job、Stage以及Task三个概念：

job：以 action 方法为界，一个 action 触发一个 job

stage：它是 job 的子集，以 RDD 宽依赖为界，遇到宽依赖即划分 stage

task：它是 stage 的子集，以分区数来衡量，分区数多少，task 就有多少
在这里插入图片描述

1.2 任务调度

spark 任务从发起到执行可用下图表示
在这里插入图片描述

1.3 Client—>ResourceManage

(1). Client 端通过 spark-submit + 参数 发起任务，即向ResourceManage 提交 application，注意该 application 包含了一堆参数，如 Executor 数，Executor 内存，Driver 内存等；

(2). ResourceManage 需要先判断现在资源是否能满足该 application，如果满足，则响应该 application，如果不满足，报错；

(3). 如果资源满足，Client 端准备 ApplicationMaster 的启动上下文，并交给 ResourceManage；

(4). 并且循环监控 application 的状态；

1.4 ResourceManage—>ApplicationMaster

(1). ResourceManage 找一个 worker 启动 ApplicationMaster；

(2). ApplicationMaster 向 ResourceManage 申请 Container；

(3). ResourceManage 收集可用资源，并告诉 ApplicationMaster；

(4). ApplicationMaster 尝试在对应的 Container 上启动 Executor 进程；

1.5 ApplicationMaster-Driver

(1). 有了资源，ApplicationMaster 启动 Driver；

//Driver 线程主要是初始化 SparkContext 对象，准备运行所需上下文，并保持与 ApplicationMaster 的 RPC 连接，通过 ApplicationMaster 申请资源

(2). Driver 启动成功后，告诉 ApplicationMaster；

1.6 Driver-Executor

(1). Executor 启动成功后，反向注册到 Driver 上，并持续向 Driver 发送心跳；

(2). Driver 启动 task，分发给 Executor，并监控 task 状态；

(3). 当 Executor 任务执行完毕后，将任务状态发送给 Driver；

spark 的核心就是资源申请和任务调度，主要通过 ApplicationMaster、Driver、Executor 来完成

spark 任务调度分为两层，一层是 stage 级的调度，一层是 task 级的调度
在这里插入图片描述

RDD 间的血缘关系，代表了计算的流程，构成了 有向无环图，即 DAG；

最后通过 action 触发 job 并调度执行；

DAGScheduler 负责 stage 级的调度，主要是将 DAG 切分成多个 stage，并将 stage 打包成 TaskSet 交给 TaskScheduler；

TaskScheduler 负责 task 级的调度，将 DAGScheduler 发过来的 TaskSet 按照指定的调度策略发送给 Executor；

SchedulerBackend 负责给 调度策略 提供可用资源，调度策略决定把 task 发送给哪个 Executor；【其中 SchedulerBackend 有多种实现，分别对接不同的资源管理系统】

基于上述认知，再来看一张图
在这里插入图片描述

Driver 在启动过程中，除了初始化 SparkContext 外，也初始化了 DAGScheduler、TaskScheduler、 SchedulerBackend 3个调度对象，同时初始化了 HeartbeatReceiver 心跳接收器；

并且各个线程之间保存通信；

SchedulerBackend 向 ApplicationMaster 申请资源，并不间断地从 TaskScheduler 获取 task 并发送给 合适的 Executor；

HeartbeatReceiver 负责接收 Executor 心跳报文，监控 Executor 存活状态；

2. Spark Stage级调度

Spark的任务调度是从DAG切割开始，主要是由DAGScheduler来完成。当遇到一个Action操作后就会触发一个Job的计算，并交给DAGScheduler来提交，下图是涉及到Job提交的相关方法调用流程图。
在这里插入图片描述

  1. Job由最终的RDD和Action方法封装而成；

  2. SparkContext将Job交给DAGScheduler提交，它会根据RDD的血缘关系构成的DAG进行切分，将一个Job划分为若干Stages，具体划分策略是，由最终的RDD不断通过依赖回溯判断父依赖是否是宽依赖，即以Shuffle为界，划分Stage，窄依赖的RDD之间被划分到同一个Stage中，可以进行pipeline式的计算。划分的Stages分两类，一类叫做ResultStage，为DAG最下游的Stage，由Action方法决定，另一类叫做ShuffleMapStage，为下游Stage准备数据，下面看一个简单的例子WordCount。
    在这里插入图片描述

Job由saveAsTextFile触发，该Job由RDD-3和saveAsTextFile方法组成，根据RDD之间的依赖关系从RDD-3开始回溯搜索，直到没有依赖的RDD-0，在回溯搜索过程中，RDD-3依赖RDD-2，并且是宽依赖，所以在RDD-2和RDD-3之间划分Stage，RDD-3被划到最后一个Stage，即ResultStage中，RDD-2依赖RDD-1，RDD-1依赖RDD-0，这些依赖都是窄依赖，所以将RDD-0、RDD-1和RDD-2划分到同一个Stage，形成pipeline操作，。即ShuffleMapStage中，实际执行的时候，数据记录会一气呵成地执行RDD-0到RDD-2的转化。不难看出，其本质上是一个深度优先搜索（Depth First Search）算法。

一个Stage是否被提交，需要判断它的父Stage是否执行，只有在父Stage执行完毕才能提交当前Stage，如果一个Stage没有父Stage，那么从该Stage开始提交。Stage提交时会将Task信息（分区信息以及方法等）序列化并被打包成TaskSet交给TaskScheduler，一个Partition对应一个Task，另一方面TaskScheduler会监控Stage的运行状态，只有Executor丢失或者Task由于Fetch失败才需要重新提交失败的Stage以调度运行失败的任务，其他类型的Task失败会在TaskScheduler的调度过程中重试。

相对来说DAGScheduler做的事情较为简单，仅仅是在Stage层面上划分DAG，提交Stage并监控相关状态信息。TaskScheduler则相对较为复杂，下面详细阐述其细节。

3. Spark Task级调度

Spark Task的调度是由TaskScheduler来完成，由前文可知，DAGScheduler将Stage打包到交给TaskScheTaskSetduler，TaskScheduler会将TaskSet封装为TaskSetManager加入到调度队列中，TaskSetManager结构如下图所示。
在这里插入图片描述

TaskSetManager负责监控管理同一个Stage中的Tasks，TaskScheduler就是以TaskSetManager为单元来调度任务。

前面也提到，TaskScheduler初始化后会启动SchedulerBackend，它负责跟外界打交道，接收Executor的注册信息，并维护Executor的状态，所以说SchedulerBackend是管“粮食”的，同时它在启动后会定期地去“询问”TaskScheduler有没有任务要运行，也就是说，它会定期地“问”TaskScheduler“我有这么余粮，你要不要啊”，TaskScheduler在SchedulerBackend“问”它的时候，会从调度队列中按照指定的调度策略选择TaskSetManager去调度运行，大致方法调用流程如下图所示：
在这里插入图片描述

上图中，将TaskSetManager加入rootPool调度池中之后，调用SchedulerBackend的riviveOffers方法给driverEndpoint发送ReviveOffer消息；driverEndpoint收到ReviveOffer消息后调用makeOffers方法，过滤出活跃状态的Executor（这些Executor都是任务启动时反向注册到Driver的Executor），然后将Executor封装成WorkerOffer对象；准备好计算资源（WorkerOffer）后，taskScheduler基于这些资源调用resourceOffer在Executor上分配task。

3.1 调度策略

TaskScheduler支持两种调度策略，一种是FIFO，也是默认的调度策略，另一种是FAIR。在TaskScheduler初始化过程中会实例化rootPool，表示树的根节点，是Pool类型。

(1) FIFO调度策略

如果是采用FIFO调度策略，则直接简单地将TaskSetManager按照先来先到的方式入队，出队时直接拿出最先进队的TaskSetManager，其树结构如下图所示，TaskSetManager保存在一个FIFO队列中。
在这里插入图片描述

(2) FAIR调度策略

FAIR调度策略的树结构如下图所示：
在这里插入图片描述

FAIR模式中有一个rootPool和多个子Pool，各个子Pool中存储着所有待分配的TaskSetMagager。

在FAIR模式中，需要先对子Pool进行排序，再对子Pool里面的TaskSetMagager进行排序，因为Pool和TaskSetMagager都继承了Schedulable特质，因此使用相同的排序算法。

排序过程的比较是基于Fair-share来比较的，每个要排序的对象包含三个属性: runningTasks值（正在运行的Task数）、minShare值、weight值，比较时会综合考量runningTasks值，minShare值以及weight值。

注意，minShare、weight的值均在公平调度配置文件fairscheduler.xml中被指定，调度池在构建阶段会读取此文件的相关配置。

  1. 如果A对象的runningTasks大于它的minShare，B对象的runningTasks小于它的minShare，那么B排在A前面；（runningTasks比minShare小的先执行）

  2. 如果A、B对象的runningTasks都小于它们的minShare，那么就比较runningTasks与minShare的比值（minShare使用率），谁小谁排前面；（minShare使用率低的先执行）

  3. 如果A、B对象的runningTasks都大于它们的minShare，那么就比较runningTasks与weight的比值（权重使用率），谁小谁排前面。（权重使用率低的先执行）

  4. 如果上述比较均相等，则比较名字。

整体上来说就是通过minShare和weight这两个参数控制比较过程，可以做到让minShare使用率和权重使用率少（实际运行task比例较少）的先运行。

FAIR模式排序完成后，所有的TaskSetManager被放入一个ArrayBuffer里，之后依次被取出并发送给Executor执行。

从调度队列中拿到TaskSetManager后，由于TaskSetManager封装了一个Stage的所有Task，并负责管理调度这些Task，那么接下来的工作就是TaskSetManager按照一定的规则一个个取出Task给TaskScheduler，TaskScheduler再交给SchedulerBackend去发到Executor上执行。

3.2 本地化调度

DAGScheduler切割Job，划分Stage, 通过调用submitStage来提交一个Stage对应的tasks，submitStage会调用submitMissingTasks，submitMissingTasks 确定每个需要计算的 task 的preferredLocations，通过调用getPreferrdeLocations()得到partition 的优先位置，由于一个partition对应一个Task，此partition的优先位置就是task的优先位置，对于要提交到TaskScheduler的TaskSet中的每一个Task，该task优先位置与其对应的partition对应的优先位置一致。

从调度队列中拿到TaskSetManager后，那么接下来的工作就是TaskSetManager按照一定的规则一个个取出task给TaskScheduler，TaskScheduler再交给SchedulerBackend去发到Executor上执行。前面也提到，TaskSetManager封装了一个Stage的所有Task，并负责管理调度这些Task。

根据每个Task的优先位置，确定Task的Locality级别，Locality一共有五种，优先级由高到低顺序：
在这里插入图片描述

在调度执行时，Spark调度总是会尽量让每个task以最高的本地性级别来启动，当一个task以X本地性级别启动，但是该本地性级别对应的所有节点都没有空闲资源而启动失败，此时并不会马上降低本地性级别启动而是在某个时间长度内再次以X本地性级别来启动该task，若超过限时时间则降级启动，去尝试下一个本地性级别，依次类推。

可以通过调大每个类别的最大容忍延迟时间，在等待阶段对应的Executor可能就会有相应的资源去执行此task，这就在在一定程度上提到了运行性能。

3.3 失败重试与黑名单机制

除了选择合适的Task调度运行外，还需要监控Task的执行状态，前面也提到，与外部打交道的是SchedulerBackend，Task被提交到Executor启动执行后，Executor会将执行状态上报给SchedulerBackend，SchedulerBackend则告诉TaskScheduler，TaskScheduler找到该Task对应的TaskSetManager，并通知到该TaskSetManager，这样TaskSetManager就知道Task的失败与成功状态，对于失败的Task，会记录它失败的次数，如果失败次数还没有超过最大重试次数，那么就把它放回待调度的Task池子中，否则整个Application失败。

在记录Task失败次数过程中，会记录它上一次失败所在的Executor Id和Host，这样下次再调度这个Task时，会使用黑名单机制，避免它被调度到上一次失败的节点上，起到一定的容错作用。黑名单记录Task上一次失败所在的Executor Id和Host，以及其对应的“拉黑”时间，“拉黑”时间是指这段时间内不要再往这个节点上调度这个Task了。

4. 总结

本图有助于理解job，stage，task工作的原理。Spark通用运行流程图，体现了基本的Spark应用程序在部署中的基本提交流程。
在这里插入图片描述

流程按照如下的核心步骤进行工作的：

  1. 任务提交后，都会先启动Driver程序；

  2. 随后Driver向集群管理器注册应用程序；

  3. 之后集群管理器根据此任务的配置文件分配Executor并启动；

  4. Driver开始执行main函数，Spark查询为懒执行，当执行到Action算子时开始反向推算，根据宽依赖进行Stage的划分，随后每一个Stage对应一个Taskset，Taskset中有多个Task，查找可用资源Executor进行调度；

  5. 根据本地化原则，Task会被分发到指定的Executor去执行，在任务执行的过程中，Executor也会不断与Driver进行通信，报告任务运行情况。

这篇关于Spark任务调度机制论述(转载)的文章就介绍到这儿，希望我们推荐的文章对编程师们有所帮助！


http://www.chinasem.cn/article/1069465。

相关文章

JVM 的类初始化机制

JVM 的类初始化机制

前言 当你在 Java 程序中new对象时，有没有考虑过 JVM 是如何把静态的字节码（byte code）转化为运行时对象的呢，这个问题看似简单，但清楚的同学相信也不会太多，这篇文章首先介绍 JVM 类初始化的机制，然后给出几个易出错的实例来分析，帮助大家更好理解这个知识点。 JVM 将字节码转化为运行时对象分为三个阶段，分别是：loading 、Linking、initialization
阅读更多...
Java ArrayList扩容机制 （源码解读）

Java ArrayList扩容机制 （源码解读）

结论：初始长度为10，若所需长度小于1.5倍原长度，则按照1.5倍扩容。若不够用则按照所需长度扩容。 一. 明确类内部重要变量含义         1：数组默认长度         2:这是一个共享的空数组实例，用于明确创建长度为0时的ArrayList ，比如通过 new ArrayList<>(0)，ArrayList 内部的数组 elementData 会指向这个 EMPTY_EL
阅读更多...
【编程底层思考】垃圾收集机制，GC算法，垃圾收集器类型概述

【编程底层思考】垃圾收集机制，GC算法，垃圾收集器类型概述

Java的垃圾收集（Garbage Collection，GC）机制是Java语言的一大特色，它负责自动管理内存的回收，释放不再使用的对象所占用的内存。以下是对Java垃圾收集机制的详细介绍： 一、垃圾收集机制概述： 对象存活判断：垃圾收集器定期检查堆内存中的对象，判断哪些对象是“垃圾”，即不再被任何引用链直接或间接引用的对象。内存回收：将判断为垃圾的对象占用的内存进行回收，以便重新使用。
阅读更多...
【Tools】大模型中的自注意力机制

【Tools】大模型中的自注意力机制

摇来摇去摇碎点点的金黄 伸手牵来一片梦的霞光 南方的小巷推开多情的门窗 年轻和我们歌唱 摇来摇去摇着温柔的阳光 轻轻托起一件梦的衣裳 古老的都市每天都改变模样                      🎵 方芳《摇太阳》 自注意力机制（Self-Attention）是一种在Transformer等大模型中经常使用的注意力机制。该机制通过对输入序列中的每个元素计算与其他元素之间的相似性，
阅读更多...
如何通俗理解注意力机制？

如何通俗理解注意力机制？

1、注意力机制（Attention Mechanism）是机器学习和深度学习中一种模拟人类注意力的方法，用于提高模型在处理大量信息时的效率和效果。通俗地理解，它就像是在一堆信息中找到最重要的部分，把注意力集中在这些关键点上，从而更好地完成任务。以下是几个简单的比喻来帮助理解注意力机制： 2、寻找重点：想象一下，你在阅读一篇文章的时候，有些段落特别重要，你会特别注意这些段落，反复阅读，而对其他部分
阅读更多...
【Tools】大模型中的注意力机制

【Tools】大模型中的注意力机制

摇来摇去摇碎点点的金黄 伸手牵来一片梦的霞光 南方的小巷推开多情的门窗 年轻和我们歌唱 摇来摇去摇着温柔的阳光 轻轻托起一件梦的衣裳 古老的都市每天都改变模样                      🎵 方芳《摇太阳》 在大模型中，注意力机制是一种重要的技术，它被广泛应用于自然语言处理领域，特别是在机器翻译和语言模型中。 注意力机制的基本思想是通过计算输入序列中各个位置的权重，以确
阅读更多...
Spark MLlib模型训练—聚类算法 PIC(Power Iteration Clustering)

Spark MLlib模型训练—聚类算法 PIC(Power Iteration Clustering)

Spark MLlib模型训练—聚类算法 PIC(Power Iteration Clustering) Power Iteration Clustering (PIC) 是一种基于图的聚类算法，用于在大规模数据集上进行高效的社区检测。PIC 算法的核心思想是通过迭代图的幂运算来发现数据中的潜在簇。该算法适用于处理大规模图数据，特别是在社交网络分析、推荐系统和生物信息学等领域具有广泛应用。Spa
阅读更多...
FreeRTOS内部机制学习03（事件组内部机制）

FreeRTOS内部机制学习03（事件组内部机制）

文章目录 事件组使用的场景事件组的核心以及Set事件API做的事情事件组的特殊之处事件组为什么不关闭中断xEventGroupSetBitsFromISR内部是怎么做的？ 事件组使用的场景 学校组织秋游，组长在等待： 张三：我到了 李四：我到了 王五：我到了 组长说：好，大家都到齐了，出发！ 秋游回来第二天就要提交一篇心得报告，组长在焦急等待：张三、李四、王五谁先写好就交谁的
阅读更多...
UVM：callback机制的意义和用法

UVM：callback机制的意义和用法

1. 作用         Callback机制在UVM验证平台，最大用处就是为了提高验证平台的可重用性。在不创建复杂的OOP层次结构前提下，针对组件中的某些行为，在其之前后之后，内置一些函数，增加或者修改UVM组件的操作，增加新的功能，从而实现一个环境多个用例。此外还可以通过Callback机制构建异常的测试用例。 2. 使用步骤         （1）在UVM组件中内嵌callback函
阅读更多...
Smarty模板引擎工作机制(一)

Smarty模板引擎工作机制(一)

深入浅出Smarty模板引擎工作机制，我们将对比使用smarty模板引擎和没使用smarty模板引擎的两种开发方式的区别，并动手开发一个自己的模板引擎，以便加深对smarty模板引擎工作机制的理解。 在没有使用Smarty模板引擎的情况下，我们都是将PHP程序和网页模板合在一起编辑的，好比下面的源代码： <?php$title="深处浅出之Smarty模板引擎工作机制";$content=
阅读更多...

Copyright China编程 23002807@qq.com

沪ICP备2023033072号-2