第3章 Spark4.x 部署模式、资源模型与存量迁移

/* MD / 富文本中的 .toc（含博客园搬家等嵌套结构）；.toc-box 在侧栏，不受影响 */
#content_views .toc,
/* 编辑器常在目录前后插入空 p（:empty 仍占 20px），一并去掉避免顶空隙 */
#content_views.markdown_views > p:empty:has(+ .toc),
#content_views.markdown_views > .toc + p:empty,
/* 富文本旧版目录标记 */
#content_views.htmledit_views #main-toc,
#content_views.htmledit_views #hr-toc,
#content_views.htmledit_views p[id*=\”-toc\”] {
display: none !important;
}
/* 目录去掉后，紧跟的首个标题不再多出一块上边距 */
#content_views.markdown_views > .toc + h1,
#content_views.markdown_views > .toc + h2,
#content_views.markdown_views > .toc + h3,
#content_views.markdown_views > .toc + h4,
#content_views.markdown_views > .toc + p:empty + h1,
#content_views.markdown_views > .toc + p:empty + h2,
#content_views.markdown_views > .toc + p:empty + h3,
#content_views.markdown_views > .toc + p:empty + h4 {
margin-top: 0 !important;
}

第3章 Spark4.x 部署模式、资源模型与存量迁移

章节定位：上一章让 Spark 能跑，这一章让它在一个已经有 Flink 的集群里跑得不出事。本章会撞见本系列第一个真正让人熬夜的问题——容器为什么会被杀掉。答案不在 Spark 的日志里，在 YARN / Kubernetes 的事件里，这也是很多人查了三小时还回头改 spark.executor.memory 的原因：改错了地方。

另一条主线是共存。把 Spark 塞进已有 Flink 的生产集群，资源争抢是明面上的问题，写入权冲突是暗处的问题，而后者会造成数据丢失而不是作业失败。

学习目标

• ✓ 能按团队现状选出 local / standalone / YARN / Kubernetes 四种模式之一，并说清 client 与 cluster 部署模式的差别到底影响什么
• ✓ 拿到 spark.executor.memory=8g，能算出容器实际向集群申请了多少、其中多少能给计算用
• ✓ 说清统一内存区 M 与不可驱逐区 R 的关系，以及为什么"缓存挤掉执行"这件事在 Spark 里方向是单向的
• ✓ 遇到"容器被杀"时，能按固定路径在 10 分钟内定位到是 overhead 不够、off-heap 没开、还是 GC 把堆撑满
• ✓ 解释动态分区分配的三个易错点，以及它和 AQE 合并分区为什么会互相抵消
• ✓ 给出 Spark 与 Flink 共集群时的资源隔离方案，以及同一张湖表的写入权仲裁机制
• ✓ 手上有一份 15 项的上线评审清单，可以拿去开会

3.1 四种部署模式与选型

表 3-1 部署模式对照

选型只有一条判断依据：你的 Flink 跑在哪，Spark 就跟着跑在哪。 两套引擎分属不同资源池，第 3.4 节所有的隔离手段都会失效——因为你无法在两个互不相干的调度器之间做仲裁。

3.1.1 client 与 cluster：不是"性能差异"，是"driver 在哪"

这是 YARN/K8s 上最容易误解的一组概念。差别只有一个：Driver 进程跑在哪里。

client 模式 cluster 模式
───────────────────────── ──────────────────────────
提交机 集群内一个容器
└─ Driver（你的终端） └─ Driver（ApplicationMaster / driver pod）
↑ ↑
│ 网络必须能直连每个 Executor │ 只需提交机连到 ResourceManager / API Server
│ （回连 driver 的 4040 端口） │
提交机下线 → 作业失联 提交机下线 → 作业继续

适用：交互式、spark-shell、Notebook 适用：所有定时批作业、生产调度

为什么生产批作业必须 cluster 模式：调度器（DolphinScheduler / CronJob）的执行节点是短生命周期的，driver 挂在它上面等于给作业加了一个"提交机不能掉线"的隐含依赖。这条在排班表上表现为"作业每天凌晨 2 点随机失败，且失败原因每次都不一样"。

K8s 上多一重麻烦：client 模式下 driver 在集群外，executor pod 要回连它，需要打通网络与鉴权。官方文档明确建议生产用 cluster 模式，本系列不提供 K8s client 模式的配置。

3.1.2 YARN 部署骨架

# conf/spark-defaults.conf 追加
spark.master yarn
spark.deploy.mode cluster
spark.yarn.submit.waitAppCompletion false # 调度器场景：提交即返回，状态交给调度器轮询

# 队列与作业标识（生产必填，第 3.4 节的隔离靠它）
spark.yarn.queue batch_normal
spark.yarn.tags owner=yangjun,team=dt,sla=T1

# AM（driver 容器）资源
spark.driver.memory 4g
spark.driver.cores 2
spark.executor.memory 8g
spark.executor.cores 4
spark.executor.memoryOverhead 2g

参数说明

验收标准：spark-submit –version 后提交一个 –class org.apache.spark.examples.SparkPi，在 YARN ResourceManager UI 看到应用状态 ACCEPTED → RUNNING → FINISHED，且 AM Node 与 Container 数量符合预期。

3.1.3 Kubernetes 部署骨架

spark-submit \\
–master k8s://https://k8s-api:6443 \\
–deploy-mode cluster \\
–conf spark.kubernetes.container.image=registry.internal/spark:4.1.4 \\
–conf spark.kubernetes.namespace=data-batch \\
–conf spark.kubernetes.authenticate.driver.serviceAccountName=spark-batch-sa \\
–conf spark.kubernetes.executor.deleteOnTermination=true \\
–conf spark.executor.memory=8g \\
–conf spark.executor.memoryOverhead=2g \\
–conf spark.kubernetes.driver.request.cores=2 \\
–conf spark.kubernetes.executor.request.cores=1 \\
–conf spark.kubernetes.executor.limit.cores=4 \\
local:///opt/spark/examples/jars/spark-examples_2.13-4.1.4.jar

参数说明（只讲与 YARN 语义不同的）

验收标准：kubectl get pods -n data-batch -w 看到 xxx-driver 先 Running，随后 executor pod 逐个出现；作业结束后 driver 进入 Completed，executor 被清理。

3.1.4 native 客户端：一个会让人困惑的形态

Spark 4 起官方提供 Spark Connect 作为默认入口，spark-submit 的行为也随之变化：它会先起一个 Connect server 再提交。在 K8s 上这带来两种镜像形态：

• 普通镜像：spark-submit 走传统路径，driver pod 内含 Spark 运行时
• Connect 服务化部署：常驻 Connect server，客户端只需 pyspark 包，不需要本地有 Spark 发行版

第 16 章展开服务化时再细讲。本章只需要记住一条排障事实：如果客户端与 server 的 Spark 小版本不一致，报错发生在 plan 序列化阶段，错误信息里不会出现版本字样，而是 Unrecognized / unsupported 类提示。这是 Spark Connect 时代新增的一类"看起来像 bug 其实是版本错配"。

3.2 内存模型：容器被杀的真正原因

3.2.1 先建立正确的 mental model

绝大多数人以为 spark.executor.memory 就是 executor 占用的内存。不是。 它只是 JVM 堆大小。

┌───────────────── 容器总内存（集群管理器看到的）─────────────────┐
│ │
│ ┌──────── JVM 堆 = spark.executor.memory（如 8g）────────┐ │
│ │ │ │
│ │ 300 MiB 预留（Spark 内部元数据，官方明确扣除） │ │
│ │ ┌──────── 统一区 M = (heap − 300MiB) × 0.6 ────────┐ │ │
│ │ │ 执行内存（shuffle / join / sort / aggregate） │ │ │
│ │ │ ┌──── 不可驱逐区 R = M × 0.5 ────┐ │ │ │
│ │ │ │ storage：缓存的 block │ ← 执行也不驱逐它 │ │ │
│ │ │ └────────────────────────────────┘ │ │ │
│ │ │ M 内其余：storage 与 execution 互相借用 │ │ │
│ │ └───────────────────────────────────────────────────┘ │ │
│ │ 其余 40%：用户数据结构、对象头、Spark 内部元数据、 │ │
│ │ 稀疏/超大记录的 OOM 缓冲 │ │
│ └────────────────────────────────────────────────────────┘ │
│ │ │
│ + 堆外 spark.memory.offHeap.size（默认 0，须显式开启） │
│ + spark.executor.memoryOverhead（默认 heap×0.10，下限 384m） │
└──────────────────────────────────────────────────────────────┘

图 3-1 Executor 内存分层

四个数字必须记住：

3.2.2 统一内存区的关键性质：驱逐是单向的

官方描述（原文要点）：

• 执行与存储共享一个统一区 M。一方不用时，另一方可占满全部
• 执行可以驱逐存储，但只能驱逐到总量跌破 R 为止
• 存储不能驱逐执行——官方理由是"实现上的复杂度"

这条不对称性直接决定了两类生产现象：

现象 A：加了 cache() 反而变慢，且磁盘 spill 暴增
根因：缓存块占住 M，执行内存被压到 R 之外，触发 spill
判据：Spark UI Storage 页有缓存，但 Shuffle/Sort 阶段出现 spill
处置：降低 memory.fraction（官方建议："cache fewer objects than slow down task execution"）
或去掉不必要的 cache()

现象 B：缓存被反复驱逐、重算，CPU 涨但 IO 没涨
根因：执行内存需求大，把 R 之外的缓存块全驱逐了，下次访问要重算
判据：Storage 页某 RDD 的 "Cached Size" 远小于 "Size in Memory"
处置：提高 memory.storageFraction，或扩大 executor 内存

为什么方向重要：如果以为是对称的，遇到现象 A 会去调大 memory.fraction（想让缓存更宽松），结果执行内存被压缩、spill 更多、作业更慢。调反方向是这类问题最常见的耗时来源。

3.2.3 三种"容器被杀"的区分与定位

这是本章最实用的一节。三者表象相同（executor 消失、作业失败或重跑），成因与解法完全不同。

第一步：先看是谁报的错
YARN 日志含 "Container killed by YARN for exceeding memory limits"
K8s 事件含 "OOMKilled" 或 exit code 137
Spark 自身日志含 "java.lang.OutOfMemoryError: Java heap space"
│
┌───────────────────────────┼───────────────────────────┐
▼ ▼ ▼
① 被集群管理器杀 ② 被管理器杀 ③ 被 JVM 自己杀
超出容器限额 超出容器限额 堆内不足
原因在 overhead 原因在 off-heap 原因在堆内
解法：加 memoryOverhead 解法：开 off-heap 或加 overhead 解法：加堆 / 减 task 并发

① overhead 不足（最高频）

典型触发条件：读 Parquet 用向量化读取器、用了 Python UDF、大量 native 库、或单条记录特别大（车联网里带嵌套信号数组的消息就是这种）。

# 现象
WARN YarnScheduler: Lost executor 3 on node5: Container on node5 ...
launched ... failed 137 (Exit code 137) ...
Container killed by YARN for exceeding memory limits.
9.0 GB of 9.0 GB physical memory used.

# 算账：heap 8g + 默认 overhead 0.8g = 容器 8.8g，但实际用了 9.0g → 杀
# 解法（三选一，按优先级）
spark.executor.memoryOverhead 2g # 直接加余量，最简单
spark.memory.offHeap.enabled true # 把 native 开销挪到可控的堆外配额
spark.sql.parquet.columnarReaderBatchSize 2048 # 减小向量化读取批，降峰值

② off-heap 未开

Spark 默认 不启用 堆外内存（spark.memory.offHeap.enabled=false，size 为 0）。官方文档建议：执行内存较大、且希望减少 GC 压力时启用，用 UnifiedMemoryManager 管理堆外。

# 开启后必须同步调低堆，否则容器总额不变但堆外挤占 overhead
spark.memory.offHeap.enabled true
spark.memory.offHeap.size 4g
spark.executor.memory 6g # 从 8g 降下来
spark.executor.memoryOverhead 2g
# 容器总额 = 6 + 4 + 2 = 12g（比原来 8.8g 大，必须同步调队列配额）

一条必须写进评审清单的纪律：开 off-heap 时，容器总额是三项相加，不是堆大小。只改 offHeap.size 不改 executor.memory，等于把容器需求悄悄抬高，然后被管理器杀掉——而且报错信息里不会提 off-heap 一个字。

③ JVM 堆内 OOM

与前两者本质不同：这是 Java 层的异常，Spark 日志里有完整栈。

java.lang.OutOfMemoryError: Java heap space
at org.apache.spark.util.collection.OpenHashSet…

判据与解法：

3.2.4 JDK 17 与 G1GC：4.x 的一个隐性变更

官方原文（Tuning 页）：

Since 4.0.0, Spark uses JDK 17 by default, which also makes the G1GC garbage collector the default. Note that with large executor heap sizes, it may be important to increase the G1 region size with -XX:G1HeapRegionSize.

这条对车联网场景有实际意义：G1 的 region 大小默认按堆容量的 1/2048 推算，堆越大 region 越大，但 humongous 对象的判定阈值是 region 的一半。单条遥测消息带大数组时，一个 record batch 就可能被判定为 humongous，直接进入 OldGen 并触发额外并发标记周期。

# 大堆 executor（≥ 16g）建议显式设 region size
spark.executor.memory 16g
spark.executor.extraJavaOptions -XX:+UseG1GC -XX:G1HeapRegionSize=16m -Xlog:gc*:file=gc-%t.log:time

为什么从 3.x 迁过来的团队会在这里踩坑：3.x 时代很多团队用 JDK 8/11 + ParallelGC，把 -XX:+UseParallelGC 写死在 extraJavaOptions 里。迁到 4.x 后这行配置仍在生效并覆盖默认，于是你以为在用 G1，其实还在用 Parallel。

# 自查：确认实际生效的 GC
spark.executor.extraJavaOptions 里有没有 UseParallelGC / UseConcMarkSweepGC / ZGC 字样
# 或运行时确认
jcmd <pid> VM.flags | grep -oE 'Use[A-Za-z]*GC'

3.3 并行度与资源分配

3.3.1 三个不同层级的"并行度"，别混为一谈

这是新手到中级最容易混淆的一组概念。Spark 里"并行度"这个词至少指三件事：

三者的关系：

可用 task 槽位总数 = executor.instances × executor.cores
↑ 这是"并发上限"

待跑 task 总数 = shuffle.partitions（shuffle 后各 stage）
↑ 这是"工作量"

工作量 > 槽位数 → 排队，正常
工作量 < 槽位数 → executor 空转，浪费钱
单 task 数据量 = 总数据量 / shuffle.partitions
↑ 这个值决定会不会 spill，通常目标区间 128MB ~ 200MB

为什么 spark.executor.cores 不是越大越好：堆大小是 executor 级的，不随 cores 增加。8g / 4cores 时每个 task 名义上能用的执行内存约为 8g × 0.6 / 4 ≈ 1.2g；改成 8g / 8cores 就变成约 600MB。同样的配置，只是"提高并发"，就把本来不 spill 的作业变成了大量小文件 spill 的作业。 这类改动在 Spark UI 上表现为 stage 耗时不降反升，且 disk spilled 从 0 变成几 GB。

3.3.2 shuffle.partitions 与 AQE 的配合

4.x 里 AQE 默认开启，coalescePartitions 会在运行时把过小的分区合并。这带来一条与 3.x 时代相反的操作建议：

可以设大，不必设小。 设大的代价是合并前多一轮调度开销（AQE 合并是在 driver 侧基于已收集的 map statistics 做的），设小的代价是真实的数据倾斜与 spill。

spark.sql.adaptive.enabled true # 4.x 默认 true
spark.sql.adaptive.coalescePartitions.enabled true # 默认 true
spark.sql.adaptive.coalescePartitions.minPartitionNum 1 # 默认即 1
spark.sql.adaptive.advisoryPartitionSizeInBytes 64MB # 合并目标大小
spark.sql.shuffle.partitions 400 # 刻意设大，交给 AQE 收

参数说明

验收标准：作业跑完后在 Spark UI 的 SQL tab 里看最终 stage 的 task 数。若等于 shuffle.partitions 说明 AQE 没合并（数据量确实大）；若明显小于它，说明合并生效。 两者都正常，但要知道是哪一种——这决定了下次数据量翻倍时会不会突然变慢。

3.3.3 动态分区分配的三个易错点

spark.dynamicAllocation.enabled=true 让 executor 数量随负载伸缩。听起来全是好处，实际有三个反复出现的坑。

易错点 1：minExecutors 会让缩容失效

spark.dynamicAllocation.minExecutors 20
spark.dynamicAllocation.maxExecutors 200

minExecutors 是下限，作业运行期间这 20 个 executor 不会被释放。设成 20 意味着：即使只剩最后一个 task 在跑，你仍然占着 20 个 executor 的配额。

官方语义：minExecutors 默认值等于 initialExecutors。也就是说你设了 initialExecutors 而没设 minExecutors 时，前者就成了缩容的下限。这是"开了动态分配但从来没缩过容"的第一大原因。

易错点 2：没有 External Shuffle Service，动态分配根本不能开

executor 内存的 shuffle 输出如果被回收就丢了。因此动态分配必须配合外部 shuffle 服务：

spark.dynamicAllocation.enabled true
spark.shuffle.service.enabled true # YARN：NM 上必须部署 ExternalShuffleService
spark.dynamicAllocation.shuffleTracking.enabled false

在 K8s 上没有 shuffle service 这个进程，改用 shuffle tracking（由 driver 跟踪哪些 executor 还持有未消费的 shuffle 数据）或持久化 shuffle 服务：

# K8s 路线：driver 跟踪未消费 shuffle 的 executor，不依赖外部服务
spark.dynamicAllocation.enabled true
spark.dynamicAllocation.shuffleTracking.enabled true

# 若要把 shuffle 数据落到外部持久化存储（跨作业复用、缩容不丢），
# 需引入 Remote Shuffle Service 并设置其插件类，例如 Celeborn 形态：
# spark.shuffle.sort.io.plugin.class = <对应发行版的 RemoteShuffleManager 插件类>
# 本系列不展开 RSS（属可选组件，且插件类名随实现不同）。
# ★ 注意：插件类全名请以所选实现的官方文档为准，本系列不给出未经取证的类名。

一条纪律：shuffleTracking 与 shuffle.service 不能同时开，且各自的适用集群管理器不同。在 YARN 上开 shuffleTracking 而在 K8s 上开 shuffle.service，都是配了不生效、但报错很晚才出现的典型错配。

易错点 3：动态分配与 AQE 合并分区会互相抵消

AQE 合并分区 → 最终 stage 的 task 数变少
↓
executor 空闲超过 executorIdleTimeout（默认 60s）→ 被回收
↓
下一个 stage task 数很多 → 重新申请 executor → JVM 启动 + 反序列化 + 注册
↓
"抖动"：executor 数在 min/max 之间反复横跳，吞吐反而下降

判据：Spark UI 的 Executors 页看 executor 的 Removal Reason 分布。大量 idle 且时间点密集交替，就是这个模式。

处置（按优先级）：

spark.dynamicAllocation.executorIdleTimeout 300s # 从默认 60s 放大，吸收 stage 间空窗
spark.dynamicAllocation.cachedExecutorIdleTimeout 1800s # 持有缓存 block 的 executor 单独放宽
spark.dynamicAllocation.minExecutors 10 # 设一个真实下限，避免缩到 0 再冷启动

3.3.4 数据本地性：一个经常被高估的旋钮

spark.locality.wait 3s # 默认值，对 PROCESS/NODE/RACK 逐级等待
spark.locality.wait.node 3s
spark.locality.wait.rack 0s # 建议直接关掉

官方在 Tuning 页给出的建议是：如果集群上大部分 task 的 locality 都是 ANY，说明数据在 HDFS 上复制因子太小，或者 task 太多，而不是"该调大 locality.wait"。

车联网场景的现实判断：湖表数据在对象存储（S3/OSS）上时，"节点本地性"这个概念本身不成立——没有 HDFS 的数据块位置。此时应当：

spark.locality.wait 0s # 对象存储：等本地性纯浪费
spark.sql.files.maxPartitionBytes 256MB # 用切片大小而不是位置来控并行度

第 2 章表 2.5.1 提到 spark.sql.maxSinglePartitionBytes 在 4.0 起默认从 Long.MaxValue 改为 128m。这条与本地性一起看：4.x 的默认行为已经在朝"不依赖数据本地性、用切片控并行"的方向走，把 locality.wait 调大是逆着这个方向。

3.4 与 Flink 共存：资源与写入权

图 3-2 共集群的双层隔离

┌──────────────── 第一层：资源隔离（防抢占）────────────────┐
│ │
│ YARN: 队列 real-time(900g) │ batch_normal(400g) │
│ 流队列 capacity 固定 + maximum-capacity 不超卖 │
│ K8s: namespace flink-rt │ data-batch │
│ ResourceQuota │ PriorityClass │
│ flink=high │ spark=batch(低) │
└──────────────────────────────────────────────────────────┘

┌──────────── 第二层：写入权仲裁（防丢数据）★ 更重要 ─────────┐
│ │
│ 湖表 dws_trip_segment │
│ ├─ 常态：Flink 流写（唯一提交方） │
│ ├─ 收口窗：Spark 接管写（02:00-04:00） │
│ └─ 仲裁：外部锁 + 时间窗 + tag 锚定 │
│ │
│ 失败模式：两边的锁互不可见 → 后提交者覆盖前者 → 静默丢数 │
└──────────────────────────────────────────────────────────┘

3.4.1 第一层：资源隔离

YARN 路线

# 流作业队列：容量固定、不允许被抢占
yarn.scheduler.capacity.root.realtime.capacity = 60
yarn.scheduler.capacity.root.realtime.maximum-capacity = 70
yarn.scheduler.capacity.root.realtime.user-limit-factor = 1

# 批作业队列：可被压到低优先级，允许借用空闲资源但不越界
yarn.scheduler.capacity.root.batch_normal.capacity = 30
yarn.scheduler.capacity.root.batch_normal.maximum-capacity = 60

# Spark 侧显式落到低优先级队列
spark.yarn.queue batch_normal
spark.yarn.priority 1 # 数字小 = 优先级高（Capacity Scheduler 语义需按发行版核对）

关键约束：批队列的 maximum-capacity 必须 < 100 − 流队列 capacity 的安全余量。否则凌晨回刷任务会把流作业的容器挤掉，表现为"白天一切正常，早上 3 点实时链路延迟飙升"。

K8s 路线

# 流：高优先级 + Guaranteed QoS
apiVersion: scheduling.k8s.io/v1
kind: PriorityClass
metadata: { name: flink–realtime }
value: 1000000
preemptionPolicy: PreemptLowerPriority # 允许抢占批作业 pod
—
# 批：低优先级 + Burstable
apiVersion: scheduling.k8s.io/v1
kind: PriorityClass
metadata: { name: spark–batch }
value: 1000
preemptionPolicy: Never

spark.kubernetes.namespace data-batch
spark.kubernetes.driver.podTemplateFile conf/spark-driver-batch.yaml
spark.kubernetes.executor.podTemplateFile conf/spark-executor-batch.yaml
# 两个 template 里设：priorityClassName: spark-batch
# requests.memory == limits.memory（避免被驱逐）

K8s 特有的驱逐风险：executor pod 若只设 requests 不设 limits，节点内存压力时它比 Flink 的 Guaranteed pod 先被驱逐。表现为"Spark 作业随机丢 executor、重跑 task"，而 YARN 上不会有这个现象。

3.4.2 第二层：写入权仲裁（比资源重要）

资源争抢的后果是慢，写入权冲突的后果是数据丢失。后者严重得多，而且不留错误。

为什么 Spark 侧特别危险：批作业失败重跑是常态。一个正在跑的 Flink 流写作业，被一个"重试了 3 次的 Spark 回刷作业"提交了新快照，Flink 侧的下一个 checkpoint 会基于错误的基线继续——没有任何一方会报错。

三层仲裁，按顺序落地：

① 外部锁（必须）
引擎自管的锁只覆盖自己。Flink 的锁看不见 Spark，反之亦然。
→ Iceberg 配 JDBC / ZK 锁；Paimon 侧靠单一提交方约定

② 时间窗（约定）
Spark 只在 02:00-04:00 持有写入权；窗口外一律只读
→ 窗口由调度器保证，不由作业自己判断（作业可能超时跨窗）

③ tag 锚定 + 原子切换（技术兜底）
接管前：给表打 tag takeover-<jobid>，作为回滚点
接管中：Flink 侧结果写影子表，不直接写目标表
结束时：Spark 完成 → 对账通过 → 原子切换 → Flink 恢复直写

Iceberg 外部锁配置示例

spark.sql.catalog.iceberg.catalog-impl org.apache.iceberg.jdbc.JdbcCatalog
spark.sql.catalog.iceberg.property.dialect default
# 关键：锁表与元数据库同库，两个引擎必须指向同一 URI
spark.sql.catalog.iceberg.uri jdbc:mysql://mysql-host:3306/iceberg_meta
spark.sql.catalog.iceberg.lock-type zookeeper
spark.sql.catalog.iceberg.lock-zookeeper-address zk1:2181,zk2:2181,zk3:2181

Flink 侧必须完全一致：

# Flink SQL
CREATE CATALOG iceberg WITH (
'type' = 'iceberg',
'catalog–type' = 'jdbc',
'uri' = 'jdbc:mysql://mysql–host:3306/iceberg_meta', –– 与 Spark 同一个
'lock.type' = 'zookeeper',
'lock.zookeeper.address' = 'zk1:2181,zk2:2181,zk3:2181' –– 与 Spark 同一个
);

验收标准（这条必须实测，不能靠 review）

测试步骤：
1. 起一个 Flink 流写作业，持续写 dws_trip_segment
2. 同时起一个 Spark 批作业，对同一张表执行 INSERT
3. 观察是否有一方收到 commit 冲突异常

预期（锁生效）：后到者抛 CommitFailedException 并重试，或等待锁
实际（锁未生效）：两边都"成功"，但其中一方的数据消失
→ 用 SELECT count(*) FROM t.snapshots 看两个快照的 parent 是否分叉

如果这个测试没通过，不要上生产。 前作在 Flink + Kafka Connect 场景记录过同类问题，结论是"同一张表不可同时由两个独立提交方写入，必须影子表隔离"。Spark 加入后，这个约束的触发概率显著上升——因为批作业的重试是设计的一部分。

3.4.3 与 Flink 的语义对照（避免拿错地图）

3.5 History Server、事件日志与作业归档

3.5.1 为什么这是第 3 章而不是第 17 章

第 17 章讲可观测与调优，但事件日志必须在搭环境的当天就打开。原因很硬：

事件日志不可追溯生成。
没开 → 那次作业的所有 Stage/Task/Shuffle/Spill 细节永久丢失。
第 17 章的 Spark UI 复盘能力，100% 建立在这份日志上。

3.5.2 配置

# 写侧（每个作业，放 spark-defaults.conf）
spark.eventLog.enabled true
spark.eventLog.dir hdfs:///spark-history # 或 s3a://bucket/spark-history
spark.eventLog.compress true # 生产建议开（lz4 默认）
spark.eventLog.logStageExecutorMetrics true # Stage 级 executor 指标
spark.eventLog.logStageTaskMetrics true

# 读侧（History Server 独立进程）
spark.history.fs.logDirectory hdfs:///spark-history
spark.history.ui.port 18080
spark.history.fs.cleaner.enabled true
spark.history.fs.cleaner.maxAge 30d
spark.history.fs.numReplayThreads 8

参数说明

3.5.3 目录规划：一个被忽视的成本

hdfs:///spark-history/
├── done/ ← 已结束作业，History Server 从这里读
└── in-progress/ ← 运行中作业（需开 spark.history.retainedApplications）

spark.history.fs.logDirectory hdfs:///spark-history/done
spark.eventLog.dir hdfs:///spark-history/in-progress
spark.history.fs.update.interval 10s

为什么分两个目录：History Server 只有作业结束后才解析 done/。若两目录合一，运行中的作业在 UI 上不可见（或显示不完整）。分开后，运行中作业看 live UI（driver 的 4040），结束后看 History Server，两者数据一致。

driver 的 4040 端口在 cluster 模式下在容器里，需要 YARN proxy 或 kubectl port-forward 才能访问。这是"cluster 模式看不到实时 UI"的真实原因，不是配置错了。

3.5.4 事件日志的版本兼容性（一个会咬人的细节）

事件日志格式随 Spark 大版本演进。Spark 4 的 History Server 能读 3.5 的日志，但反过来不行（旧 server 读到新 schema 会 replay 失败）。

迁移期的正确顺序：
1. 先升级 History Server 到 4.x（向后兼容读旧日志）
2. 再逐个升级作业（产生新格式日志）
3. 保留 3.5 时期的 done/ 目录，不删除
错误顺序：先升作业后升 server → 历史 UI 整片打不开，且报错是 replay 异常，
看起来像日志损坏，容易误删目录（不可恢复）

验收标准

# 1. 写侧确认日志生成
hdfs dfs -ls /spark-history/in-progress | head
# 预期：每个运行中作业一个文件，大小随时间增长

# 2. 结束后确认归档
hdfs dfs -ls /spark-history/done | head
# 预期：作业 FINISHED 后文件出现在此目录

# 3. History Server 可见
curl -s http://history-host:18080/json/version | head
# 预期：返回 JSON，且 UI 列表能看到刚完成的作业

3.6 生产上线 checklist（15 项）

这一节可以直接打印带去评审会。每项都是"不满足就不该上线"的级别。

第 8 项是这张表里唯一"后果是丢数据"而非"后果是变慢/失败"的项。如果时间只够验一项，验它。

工程要点

本章小结

本章解决"跑在哪、给多少、怎么和 Flink 共处"。

• 部署模式（3.1）：判断依据只有一条——Flink 在哪 Spark 就在哪。生产批作业一律 cluster 模式，因为 driver 挂在提交机上等于加了一个"不能掉线"的隐含依赖。
• 内存模型（3.2）：容器 = 堆 + off-heap + overhead 这道加法，是全章的支点。统一区 M 与不可驱逐区 R 的单向性，决定了"加缓存变慢"该往哪个方向调。三种容器被杀的区分路径（管理器杀-overhead / 管理器杀-offheap / JVM 自杀）是本章最实用的产出。
• 并行度（3.3）：三个层级（instances / cores / partitions）不能混为一谈；AQE 时代 shuffle.partitions 可以设大不必设小；动态分配的三个坑各有明确判据。
• 共存（3.4）：双层隔离，第二层（写入权仲裁）比第一层（资源）重要，因为它的失败模式是静默丢数据。三层仲裁（外部锁 + 时间窗 + tag 锚定）必须实测。
• 可观测地基（3.5）：事件日志当天开、目录分 done/in-progress、History Server 先升级。

15 项上线清单里，第 8 项是唯一"后果为丢数据"的——如果时间只够验一项，验它。

下章预告：底座稳了，开始讲内核。第 4 章《Catalyst、执行计划与 AQE》——一条 SQL 从文本到物理计划要过哪四段变换、Whole-Stage CodeGen 为什么能把逐行处理编译成一段 Java、EXPLAIN 的五种输出各自回答什么问题、AQE 的四个开关怎么确认"真的生效了"而不是"配了但没用"，以及一套跨引擎的读计划方法：Spark 的 EXPLAIN、ClickHouse 的 EXPLAIN indexes = 1、Doris 的 EXPLAIN，三套地图怎么对照着看。学完这章，你拿到一条 SQL 应该能在三分钟内判断它会不会慢、慢在哪。

官方参考资料

版本取证说明

本章遗留待办：configuration.html 与 sql-performance-tuning.html 的默认值本次未逐字抓取，上表已标注。第 17 章（性能调优）动笔前必须补齐逐字复核——那一章会大量引用这些默认值。

实操清单

本章为部署章，以下每项可复现。第 4、8 项是本章的核心实验。

思考题

下一章进入内核。前三章解决了"该不该用、跑在哪、给多少"，从这里开始回答那个更本质的问题：一条 SQL 交给 Spark，它到底读了多少数据、为什么是这个数、以及怎么让它少读十倍。