动态资源分配与作业调度：多租户集群的资源博弈

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

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

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

动态资源分配（DRA）

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

作业生命周期中的 Executor 数量变化:

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

  Stage 0 有 8 个 Task → 请求 8 个 Executor
  Stage 0 完成，等待 Stage 1 → 空闲 Executor 逐步释放
  Stage 1 有 4 个 Task → 保留 4 个 Executor
  作业结束 → 所有 Executor 释放

DRA 的核心配置：

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 有一个前置条件：必须部署 External Shuffle Service（ESS）。

原因是：当一个 Executor 空闲被释放时，它本地磁盘上的 Shuffle 文件也随之消失。如果下游 Stage 还需要读取这些 Shuffle 数据，就会触发 FetchFailedException，导致上游 Stage 重算。

ESS 把 Shuffle 文件的管理从 Executor 进程中分离出来。ESS 是一个独立的服务进程，运行在每个集群节点上。Executor 写出的 Shuffle 文件由 ESS 管理，即使 Executor 进程退出，Shuffle 文件仍然可以通过 ESS 读取。

无 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）。在 K8s 上，ESS 通常以 DaemonSet 形式部署到每个节点。

作业调度器：FIFO vs FAIR

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

FIFO 调度器（默认）：先提交的 Job 优先获得所有资源。后提交的 Job 必须等前一个 Job 的所有 Task 完成后才能开始。简单但不公平——一个大 Job 会阻塞所有后续小 Job。

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

FIFO (默认):
  Job 0 (100 Tasks): ████████████████████████████████████████████████████
  Job 1 (10 Tasks):                                                      ██
  Job 2 (5 Tasks):                                                         █
  → Job 1 必须等 Job 0 完成

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

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

<!-- 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 之间公平分配资源。

集群资源 (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 行为

// 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。如果集群资源紧张，两者同时开启可能加剧资源争抢。

练习

配置 DRA（minExecutors=1, maxExecutors=10），提交一个 100 分区的 Shuffle 作业。在 Spark UI 的 Executors 页面记录 Executor 数量随时间的变化曲线。
在同一个 SparkContext 中从两个线程同时提交两个作业（一个大一个小），分别在 FIFO 和 FAIR 模式下执行。对比小作业的完成时间。
在不部署 ESS 的情况下开启 DRA，提交一个两阶段 Shuffle 作业。观察第一阶段的 Executor 被释放后，第二阶段是否出现 FetchFailedException。

系列导航

序号	主题	状态
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：两种计算引擎的设计选择