当前位置：首页 > news >正文

Spark 运行流程核心组件（二）任务调度

news 2025/8/17 14:36:44

1、调度策略

参数	默认值	说明
`spark.scheduler.mode`	`FIFO`	调度策略（FIFO/FAIR）
`spark.locality.wait`	`3s`	本地性降级等待时间
`spark.locality.wait.process`	`spark.locality.wait`	PROCESS_LOCAL 等待时间
`spark.locality.wait.node`	`spark.locality.wait`	NODE_LOCAL 等待时间
`spark.locality.wait.rack`	`spark.locality.wait`	RACK_LOCAL 等待时间

调度策略：FIFO 按提交顺序处理；FAIR 支持权重分配。

本地化策略：从 PROCESS_LOCAL 到 ANY 逐级降级，减少数据传输开销

val PROCESS_LOCAL, NODE_LOCAL, NO_PREF, RACK_LOCAL, ANY = Value性能最优-> 最差
PROCESS_LOCAL -> NODE_LOCAL -> NO_PREF -> RACK_LOCAL -> ANY1、PROCESS_LOCAL
任务所需的数据就在同一个 Executor 进程的内存中2、NODE_LOCAL
任务所需的数据在同一个物理节点上，但不在同一个 Executor 进程的内存中。3、NO_PREF
没有位置偏好，数据来源本身是均匀分布或位置无关。4、RACK_LOCAL
不在同一个节点上，但在同一个机架5、ANY
在集群的其他机架

2、核心组件

1、Schedulable

可调度实体的接口定义

1、Pool

调度树中的节点，可以是根节点或中间节点

调度模式（FIFO或FAIR）
权重（weight）
最小资源份额（minShare）
运行任务数（runningTasks）
子节点列表（schedulableQueue）

2、TaskSetManager

管理一个TaskSet（一组任务），负责任务调度、本地性处理、失败重试和推测执行

跟踪任务状态（待运行、运行中、已完成）
根据数据本地性选择任务
处理任务失败和重试逻辑
实现推测执行机制

2、TaskScheduler

Spark任务调度的核心实现，协调资源分配和任务调度

接收DAGScheduler提交的TaskSet
管理Executor资源状态
分配任务到可用Executor
处理任务状态更新
实现调度延迟和推测执行策略

3、SchedulableBuilder

调度树构建的抽象基类

1、FIFOSchedulableBuilder

实现先进先出调度策略的构建器

2、FairSchedulableBuilder

实现公平调度策略的构建器

3、核心设计

1、调度树

在这里插入图片描述

Spark Standalone支持树形结构的调度池，每个池可以独立配置调度模式（FIFO/FAIR）

特性	FIFO调度	公平调度
结构	单层Pool结构	多层Pool树结构
排序方式	按提交顺序	基于权重/minShare
资源分配	独占式	按比例共享
适用场景	批处理作业	多用户/多作业环境
配置方式	无需配置	XML配置文件定义Pool

作业通过sc.setLocalProperty("spark.scheduler.pool", "poolName")分配到指定的调度池
配置调度池配置文件

<!-- conf/fairscheduler.xml -->
<?xml version="1.0"?>
<allocations><pool name="production"><schedulingMode>FAIR</schedulingMode><weight>1</weight><minShare>2</minShare></pool><pool name="test"><schedulingMode>FIFO</schedulingMode><weight>2</weight><minShare>3</minShare></pool>
</allocations>

1、FIFO

    // 优先级val priority1 = s1.priorityval priority2 = s2.priorityvar res = math.signum(priority1 - priority2)if (res == 0) {val stageId1 = s1.stageIdval stageId2 = s2.stageId// 先进先出res = math.signum(stageId1 - stageId2)}res < 0

2、FAIR

override def comparator(s1: Schedulable, s2: Schedulable): Boolean = {val minShare1 = s1.minShareval minShare2 = s2.minShareval runningTasks1 = s1.runningTasksval runningTasks2 = s2.runningTasks// 1. 满足minShare优先级val s1Needy = runningTasks1 < minShare1val s2Needy = runningTasks2 < minShare2// 2. 比较资源使用比例val minShareRatio1 = runningTasks1.toDouble / math.max(minShare1, 1.0)val minShareRatio2 = runningTasks2.toDouble / math.max(minShare2, 1.0)// 3. 权重比较val taskToWeightRatio1 = runningTasks1.toDouble / s1.weight.toDoubleval taskToWeightRatio2 = runningTasks2.toDouble / s2.weight.toDoublevar compare = 0if (s1Needy && !s2Needy) {return true} else if (!s1Needy && s2Needy) {return false} else if (s1Needy && s2Needy) {compare = minShareRatio1.compareTo(minShareRatio2)} else {compare = taskToWeightRatio1.compareTo(taskToWeightRatio2)}if (compare < 0) {true} else if (compare > 0) {false} else {s1.name < s2.name}}

2、资源分配与任务调度

Executor注册

executorAdded(o.executorId, o.host)

Pool排序调度队列

val sortedTaskSets = rootPool.getSortedTaskSetQueue

TaskSetManager资源分配请求

val (noDelayScheduleReject, minLocality) = resourceOfferSingleTaskSet(taskSet, currentMaxLocality, shuffledOffers, availableCpus,availableResources, tasks)val (taskDescOption, didReject, index) =taskSet.resourceOffer(execId, host, maxLocality, taskCpus, taskResAssignments)

TaskSetManager本地性匹配

var allowedLocality = maxLocalityif (maxLocality != TaskLocality.NO_PREF) {allowedLocality = getAllowedLocalityLevel(curTime)if (allowedLocality > maxLocality) {// We're not allowed to search for farther-away tasksallowedLocality = maxLocality}}