上一篇解决了 Spark 在 YARN 和 K8s 上的部署模式。这一篇进入资源调度——在多个作业共享集群时，Spark 如何动态分配和回收资源。

默认情况下，一个 Spark 应用在启动时申请固定数量的 Executor，作业结束前一直占用。这在单用户开发环境下可以接受，在多租户生产集群中会造成严重的资源浪费——一个空闲的 Spark 应用占着 100 个 Executor 不释放，其他作业排队等待。

本文只抓一个问题：动态资源分配如何按需申请和释放 Executor，以及 FAIR 调度器如何在多个作业之间公平分配资源。

图 1：动态资源分配与两级调度的责任边界

flowchart TB
  Q[集群队列或 namespace 配额] --> A[应用可申请的资源上限]
  B[等待中的 Task 与空闲超时] --> D[Dynamic Allocation]
  D -->|请求增减| E[应用专属 Executor]
  A --> E
  F[FIFO 或 FAIR：应用内部] -->|选择待运行 Task| E

图 1 中 FAIR 只调度同一 SparkContext 内的 Job。跨应用的资源分配由集群管理器决定；动态分配发出请求后，仍可能因队列容量或配额不足而得不到 Executor。

动态资源分配（DRA）

Dynamic Resource Allocation（DRA）允许 Spark 应用根据工作负载动态增减 Executor 数量：有 pending Task 时申请新 Executor，Executor 空闲超时后释放。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
作业生命周期中的 Executor 数量变化:

  Executor
  数量
   8 │          ┌────┐
   6 │     ┌────┤    ├────┐
   4 │ ┌───┤    │    │    ├───┐
   2 │ │   │    │    │    │   │
   0 │─┘   │    │    │    │   └──
     └──────────────────────────→ 时间
      启动  Stage0  Stage1  空闲  结束

  Stage 0 有 pending Task → 根据每个 Executor 的 Task 容量计算需求
  Stage 0 完成，等待 Stage 1 → 空闲 Executor 逐步释放
  Stage 1 Task 数减少 → 结合空闲超时及 Shuffle 保留条件决定缩容
  作业结束 → 所有 Executor 释放

DRA 的核心配置：

1
2
3
4
5
6
spark.dynamicAllocation.enabled=true           开启 DRA
spark.dynamicAllocation.minExecutors=1         最少保留的 Executor 数
spark.dynamicAllocation.maxExecutors=100       最多申请的 Executor 数
spark.dynamicAllocation.executorIdleTimeout=60s  空闲多久后释放
spark.dynamicAllocation.schedulerBacklogTimeout=1s  有 pending Task 多久后申请新 Executor
spark.dynamicAllocation.sustainedSchedulerBacklogTimeout=1s  持续有 pending Task 时的申请间隔

DRA 的扩容策略是指数增长：第一次申请 1 个 Executor，如果仍有 pending Task，第二次申请 2 个，第三次 4 个，以此类推。这既避免了一次申请大量 Executor 造成资源浪费，又能在持续高负载时快速扩容。

External Shuffle Service

DRA 需要一种支持安全回收 Executor 的 Shuffle 保留机制。官方列出 ESS、shuffle tracking、启用 Shuffle block 迁移的 decommission，以及提供可靠存储能力的 ShuffleDataIO 插件等选择，ESS 只是其中一种。

需要防止的是 Executor 回收后 Shuffle 输出不可访问。文件可能仍在节点磁盘上，但若没有继续提供读取的服务，也无法供下游使用。shuffle tracking 会避免过早回收仍持有需要的 Shuffle 输出的 Executor；ESS 则将读取服务与 Executor 进程分开。

ESS 把 Shuffle 文件的读取服务从 Executor 生命周期中分离出来，在节点级服务中运行。YARN 将它作为 NodeManager 的 Auxiliary Service 加载。只要服务和本地文件仍可用，即使 Executor 退出，下游仍能读取这些文件。

1
2
3
4
5
6
7
8
9
无 ESS 或其他保留机制，且输出已不可访问:
  Executor 0 → 写 Shuffle 文件到本地磁盘
  Executor 0 被 DRA 释放 → 文件丢失
  下游 Task → FetchFailed → 上游重算

有 ESS:
  Executor 0 → 写 Shuffle 文件，注册到 ESS
  Executor 0 被 DRA 释放
  下游 Task → 通过 ESS 读取 Shuffle 文件 → 正常执行

在 YARN 上，ESS 作为 YARN NodeManager 的 Auxiliary Service 部署（yarn.nodemanager.aux-services=spark_shuffle）。Kubernetes 原生集成目前不内置 ESS；不能把自行部署一个 DaemonSet 当作已受支持的配置。可按对应版本配置 shuffle tracking 或其他适用的 Shuffle 保留机制。

作业调度器：FIFO vs FAIR

当一个 SparkContext 中有多个 Job 同时运行（比如不同线程提交的多个 action），内部调度器决定 Job 之间的资源分配策略。

FIFO 调度器（默认）：先提交的 Job 优先获得资源。先提交的 Job 没有足够可运行 Task 占满资源时，后续 Job 仍可利用空闲资源；资源持续被占满时，小 Job 可能等待较久。

FAIR 调度器：所有 Job 轮流获得资源。多个 Job 的 Task 交替执行，每个 Job 都能获得部分资源。适合交互式查询场景——一个大 ETL 作业不会阻塞临时的小查询。

1
2
3
4
5
6
7
8
9
10
11
FIFO (默认):
  Job 0 (100 Tasks): ████████████████████████████████████████████████████
  Job 1 (10 Tasks):                                                      ██
  Job 2 (5 Tasks):                                                         █
  → 示例中 Job 0 占满资源；资源空闲时后续 Job 可运行

FAIR:
  Job 0 (100 Tasks): █ █ █ █ █ █ █ █ █ █ █ █ █ █ █ █ █ █ █ █ ...
  Job 1 (10 Tasks):   █ █ █ █ █ █ █ █ █ █
  Job 2 (5 Tasks):     █ █ █ █ █
  → 三个 Job 交替获得资源

FAIR 调度器还支持调度池（Scheduler Pool），可以把不同类别的 Job 分到不同的池中，每个池有独立的权重和最小资源保证：

1
2
3
4
5
6
7
8
9
10
11
<!-- fairscheduler.xml -->
<pool name="production">
  <schedulingMode>FAIR</schedulingMode>
  <weight>3</weight>
  <minShare>10</minShare>
</pool>
<pool name="adhoc">
  <schedulingMode>FIFO</schedulingMode>
  <weight>1</weight>
  <minShare>2</minShare>
</pool>

多租户场景的资源隔离

在多租户集群上，资源隔离需要在两个层面实现：

集群层面（YARN / K8s）：通过 YARN Queue 或 K8s Namespace + ResourceQuota 限制每个租户可以使用的总资源量。一个租户最多使用 500 核 CPU 和 2TB 内存。

应用层面（Spark 调度器）：通过 FAIR 调度器和调度池，在单个 Spark 应用内部的多个 Job 之间公平分配资源。

1
2
3
4
5
6
7
8
集群资源 (1000 CPU, 4TB Memory)
├── Queue A (团队 A): 最大 500 CPU
│   ├── Spark App 1: DRA, max 50 Executors
│   └── Spark App 2: DRA, max 30 Executors
└── Queue B (团队 B): 最大 500 CPU
    └── Spark App 3: FAIR 调度器
        ├── Pool "etl": weight=3
        └── Pool "query": weight=1

实验：观察 DRA 行为

1
2
3
4
5
6
7
8
9
10
11
12
13
14
// spark-shell 启动时开启 DRA
// spark-shell --conf spark.dynamicAllocation.enabled=true \
//             --conf spark.dynamicAllocation.minExecutors=1 \
//             --conf spark.dynamicAllocation.maxExecutors=10 \
//             --conf spark.dynamicAllocation.executorIdleTimeout=30s

val rdd = sc.parallelize(1 to 10000000, 100)
rdd.map(x => (x % 1000, x)).reduceByKey(_ + _).count()

// 在 Spark UI 的 Executors 页面观察:
// 1. 初始只有 1 个 Executor
// 2. 检测到 pending Tasks 后逐步增加 (1, 2, 4, 8...)
// 3. 作业完成后 Executor 逐步释放
// 4. 30 秒空闲后回到 minExecutors=1

模式提炼

DRA 体现了弹性资源管理的通用模式：按需申请、用完释放、指数扩容、延迟缩容。

同样的模式在云计算中广泛使用：AWS Auto Scaling Group 的实例伸缩、K8s 的 HPA（Horizontal Pod Autoscaler）、Serverless 函数的冷启动和自动伸缩。核心挑战都是一样的——在响应速度（快速扩容）和资源效率（及时缩容）之间找平衡。

FAIR 调度器体现了公平分享模式（Fair Share Scheduling），起源于操作系统的进程调度。YARN 的 FairScheduler、Hadoop 的 CapacityScheduler、Linux 的 CFS（Completely Fair Scheduler）都使用相同的思想。

工程迁移表

Spark 概念 YARN 对应 K8s 对应 操作系统对应
DRA（动态 Executor 分配） YARN Container 动态申请 Pod HPA / 动态 Pod 创建 进程池的动态 fork/kill
External Shuffle Service NodeManager Auxiliary Service DaemonSet 共享文件系统
FAIR Scheduler FairScheduler / CapacityScheduler 无直接对应 CFS（完全公平调度器）
调度池 YARN Queue K8s Namespace + ResourceQuota cgroup
executorIdleTimeout Container 空闲回收 Pod 空闲删除 进程超时回收

常见误解

误解一：“DRA 开启后不需要设置 maxExecutors”。如果不设 maxExecutors，一个大作业可能申请集群中所有可用资源，挤占其他作业。生产环境必须根据公平性要求设置上限。

误解二：“FAIR 调度器在作业之间分配 Executor”。FAIR 调度器分配的是 Task 执行时间片，不是 Executor。所有 Job 共享同一组 Executor，FAIR 调度器决定在这些 Executor 上优先执行哪个 Job 的 Task。

误解三：“DRA 和推测执行不能同时使用”。可以同时使用，但需要注意推测执行会创建额外的 Task，这可能触发 DRA 申请更多 Executor。如果集群资源紧张，两者同时开启可能加剧资源争抢。

练习

  1. 配置 DRA（minExecutors=1, maxExecutors=10），提交一个 100 分区的 Shuffle 作业。在 Spark UI 的 Executors 页面记录 Executor 数量随时间的变化曲线。

  2. 在同一个 SparkContext 中从两个线程同时提交两个作业（一个大一个小），分别在 FIFO 和 FAIR 模式下执行。对比小作业的完成时间。

  3. 在测试集群确认启用 shuffle tracking 后，提交一个两阶段 Shuffle 作业。观察持有仍被需要的 Shuffle 输出的 Executor 是否延迟回收，并对照不再需要输出后的回收时间。

系列导航

序号 主题 状态
00 导读：为什么 Spark 的核心是一张 DAG 已完成
01 RDD：弹性分布式数据集的五大属性 已完成
02 宽依赖与窄依赖：Stage 是怎样划分出来的 已完成
03 DAG Scheduler 与 Task Scheduler：从逻辑计划到物理执行 已完成
04 Shuffle 机制：数据跨分区交换的代价与优化 已完成
05 内存管理与 Tungsten：堆外内存、序列化与代码生成 已完成
06 存储体系：Block Manager、广播变量与累加器 已完成
07 容错机制：Lineage、Checkpoint 与推测执行 已完成
08 Spark SQL 与 Catalyst：从 SQL 文本到物理算子树 已完成
09 DataFrame 与 Dataset：类型安全与性能的折中 已完成
10 Adaptive Query Execution：运行时改写执行计划 已完成
11 数据源 API 与谓词下推：让存储层帮忙过滤 已完成
12 Structured Streaming：把流当成无界表 已完成
13 水位线与状态管理：流处理的时间语义 已完成
14 Spark on YARN / K8s：资源管理与部署模式 已完成
15 动态资源分配与作业调度：多租户集群的资源博弈 本篇
16 性能诊断：Spark UI、Event Log 与调优思路 下一篇
17 Spark 与 Flink：两种计算引擎的设计选择

参考资料