bg13.一致性入湖
kafka-flink-iceberg 加餐
| 语义 | 定义 | 失败时的行为 | 后果 |
|---|---|---|---|
| at-most-once（至多一次） | 每条数据最多被处理一次 | 失败就丢弃，不重试 | 可能丢数据，绝不重复 |
| at-least-once（至少一次） | 每条数据至少被处理一次 | 失败就重试 | 绝不丢数据，可能重复 |
| exactly-once（精确一次） | 每条数据对最终结果的影响恰好等于一次 | 失败重试，但重复的影响被消除 | 既不丢也不重 |
误解：exactly-once 意思是"这条消息在网络上只被传输了一次"。
这是错的，而且错得很关键。物理层面的重复传输一定会发生——Flink 从 checkpoint 恢复时，一定会把上次 checkpoint 之后的数据重新读一遍、重新算一遍。
正确的表述是：
exactly-once 指的是：无论底层重传、重算了多少次，这条数据对「最终对外可见的状态」的影响，恰好等于处理一次的效果。
所以业界更精确的叫法是 effectively-once（等效一次）或 exactly-once state consistency（精确一次的状态一致性）。
这个区分为什么重要？ 因为它直接决定了实现方式：既然重算无法避免，那唯一的出路就是——让重算产生的中间结果对外不可见，直到确认这一批算对了，再一次性地让它全部可见。
这句话就是本篇后面所有机制的总纲。
假设统计"每分钟的点击数"，Flink 在算完 14:05 这一分钟的结果（值为 1000）、写出去之后、提交 checkpoint 之前崩溃了。
| 语义 | 恢复后的行为 | 最终结果 |
|---|---|---|
| at-most-once | 从最新数据开始读，14:05 那一分钟的数据不再处理 | 丢失：14:05 这一分钟可能只有 300 |
| at-least-once | 从上次 checkpoint 重放，14:05 又算了一遍又写了一次 | 重复：下游看到 1000 + 1000 = 2000 |
| exactly-once | 从上次 checkpoint 重放，重算出 1000，但第一次写出的那份从未对外可见，被丢弃 | 正确：1000 |
关键在第三行的"从未对外可见"。这需要 sink（数据写出的目标端）配合——它必须支持"先写、后确认，未确认的不可见"。这就引出了两阶段提交。
端到端链路是 Kafka → Flink → Iceberg。每一段单独看都不足以保证 exactly-once，看清各自的缺口，才能理解为什么需要一个跨段的协议。
Kafka 的消费位置由 offset 标记。Flink 不使用 Kafka 自带的 offset 提交机制来控制重放，而是把 offset 存进自己的 checkpoint。
- 能保证的：恢复时能精确回到上次 checkpoint 记录的 offset，从那里重新读。数据不会丢。
- 缺口：恢复后这些数据会被重新处理一遍。Source 段本身没有能力阻止重复处理的后果。
注：Flink 也会把 offset 提交回 Kafka，但那只是为了让 Kafka 的监控工具能看到消费进度，不作为恢复依据。这是一个容易误解的点——看 Kafka 的 lag 监控和看 Flink 的 checkpoint 状态，说的不是同一件事。
checkpoint 机制 保证：所有算子的 state 都是同一个"逻辑时刻"的一致快照。
- 能保证的：Flink 内部 state 的 exactly-once。恢复后，所有算子的状态回到同一个一致点，不会出现"算子 A 已经处理了这条数据、算子 B 还没有"的撕裂。
- 缺口：管不着已经写出去的数据。Flink 无法把已经写进 Iceberg 的文件"收回来"。
这是缺口最大的一段。一个普通的（非事务的）sink 收到数据就直接写、直接可见。
- 能保证的：什么都保证不了。
- 缺口：崩溃前写出去的数据已经对外可见，恢复后重算又会再写一份。这就是重复的来源。
Kafka Flink 内部 Iceberg
│ │ │
offset 存进 state 一致快照 文件已经写出去了
checkpoint ✓ ✓ ✗ 收不回来
│ │ │
┌───┴────────────────────┴─────────────────────┴───┐
│ 三段各自的保证拼不出端到端保证: │
│ Source 能重放 + 内部能恢复 = 一定会重算 │
│ 重算 + Sink 立即可见 = 一定会重复 │
└──────────────────────────────────────────────────┘
↓
需要一个协议,让 Sink 的写入"延迟可见",
并且这个"何时可见"的决定权交给 checkpoint
↓
两阶段提交
2PC 于是乎登场：
问题：有多个互相独立的系统，要让一个操作在它们身上要么全部生效，要么全部不生效，不允许出现"一半生效"。
难点在于：每个系统都可能在任意时刻崩溃或失联，而且没有任何一个系统能看见其他系统的内部状态。
角色：
- 协调者（coordinator）：一个，负责发号施令并做最终决定
- 参与者（participant）：多个，负责执行实际操作
阶段一：准备（prepare / pre-commit）
协调者问所有参与者："你能提交吗？"
每个参与者收到后，把操作做到"只差最后一步"的程度——数据写好了、资源占住了、日志记好了，但对外还不可见。然后回答"可以"或"不行"。
关键约束：一个参与者一旦回答了"可以"，它就必须保证后续无论如何都能提交成功，哪怕它中间崩溃重启，恢复后也必须还能提交。这意味着 prepare 的结果必须持久化。
阶段二：提交（commit）或中止（abort）
- 如果所有参与者都答"可以" → 协调者广播 commit，各方把成果变为可见
- 只要有一个答"不行"或超时 → 协调者广播 abort，各方回滚
sequenceDiagram
participant C as 协调者
participant P1 as 参与者 1
participant P2 as 参与者 2
Note over C,P2: 阶段一:准备
C->>P1: prepare?
C->>P2: prepare?
P1->>P1: 写好数据,持久化,但不可见
P2->>P2: 写好数据,持久化,但不可见
P1-->>C: yes(我保证后续一定能提交)
P2-->>C: yes
Note over C,P2: 阶段二:提交
C->>C: 全部 yes → 决定 commit(此决定必须先持久化)
C->>P1: commit
C->>P2: commit
P1->>P1: 成果变为可见
P2->>P2: 成果变为可见
保证：所有参与者的最终状态一致（全部提交或全部中止）。
代价一：阻塞。 参与者在回答"yes"之后、收到最终决定之前，处于不确定状态——它不能自作主张提交（万一别人失败了），也不能中止（万一别人都成功了）。这期间它占用的资源（锁、内存、连接）无法释放。如果协调者在此时宕机，参与者会一直阻塞下去。
代价二：协调者是单点。 协调者在"决定了 commit 但还没广播完"的瞬间宕机，是最糟的情况——有的参与者提交了，有的没收到。恢复的前提是协调者把决定持久化了，重启后能继续广播。
代价三：延迟。 一次操作要两轮网络往返，且必须等最慢的参与者。
"做到只差最后一步，但对外不可见"。这就是为什么 Flink 选择用 2PC 来做端到端 exactly-once：它要的那个"延迟可见"的能力，恰好就是 2PC 的 prepare 阶段。
| 2PC 角色 | Flink 里是谁 |
|---|---|
| 协调者 | JobManager 里的 Checkpoint Coordinator（检查点协调器）。它是 JobManager 的组成部分，负责发起 checkpoint 并收集各算子的确认 |
| 参与者 | 各个 sink 算子的并行实例（跑在 TaskManager 的 task slot 里） |
| prepare 阶段 | checkpoint 过程中，sink 执行 pre-commit：把数据写成文件，但不让它进入 Iceberg 的当前快照 |
| commit 决定 | Checkpoint Coordinator 收齐所有算子的确认，宣布本次 checkpoint 完成 |
| commit 广播 | 框架回调各 sink 的 notifyCheckpointComplete 方法 |
Flink 把这套逻辑封装在 TwoPhaseCommitSinkFunction（两阶段提交 Sink 抽象类）里；新版 Sink API 中对应的是带 Committer 的两阶段提交 Sink。Iceberg、Kafka 的 Flink connector 都基于它实现。
⚠️ 撞名提醒：这里的"Checkpoint Coordinator"是 Flink JobManager 的内部组件，和 Kafka 的"消费组协调器（Group Coordinator）"、和 2PC 理论里的"协调者"是三个不同层面的东西，只是名字都叫协调者。本篇提到"协调者"时，除非特别说明，指的都是 Flink 的 Checkpoint Coordinator。
sequenceDiagram
participant JM as JobManager<br/>(Checkpoint Coordinator)
participant SRC as Source 算子
participant OP as 中间算子
participant SINK as Iceberg Sink 算子
participant S3 as 对象存储
participant CAT as Iceberg Catalog<br/>(元数据)
Note over JM,CAT: ── 正常运行:数据文件持续写出,但对读者不可见 ──
SINK->>S3: 写 part-0001.parquet
SINK->>S3: 写 part-0002.parquet
Note right of S3: 文件已在对象存储上<br/>但没被任何 snapshot 引用<br/>→ 查询看不到
Note over JM,CAT: ── 阶段一:prepare ──
JM->>SRC: 注入 checkpoint barrier(编号 N)
SRC->>SRC: 把当前 offset 存入 state
SRC->>OP: barrier N 随数据流向下游
OP->>OP: 把算子 state 存入 state backend
OP->>SINK: barrier N
SINK->>S3: 关闭当前文件,flush 落盘
SINK->>SINK: 把"这批文件的清单"作为 state 持久化<br/>(pre-commit 完成)
SINK-->>JM: ack checkpoint N
Note over JM,CAT: ── 阶段二:commit ──
JM->>JM: 收齐所有算子 ack<br/>→ 宣布 checkpoint N 完成(并持久化该决定)
JM->>SINK: notifyCheckpointComplete(N)
SINK->>CAT: 提交:把这批文件加入新 snapshot<br/>(原子操作,见 4.3)
Note right of CAT: 此刻起,查询才能看到这批数据
注意时序里的一个关键点：数据文件在 barrier 到达之前就已经在写了。pre-commit 做的不是"开始写"，而是"封口"——关闭文件、确保落盘、把文件清单记进 state。真正决定可见性的，是阶段二对 Catalog 的那次提交。
Iceberg 的表结构是这样的：
Catalog(指针) ──▶ metadata.json(当前版本) ──▶ manifest list ──▶ manifest ──▶ 数据文件.parquet
↑
"当前快照"就是这个指针指向的那份
pre-commit 阶段：sink 把 Parquet 文件写到对象存储上，并生成描述这些文件的元信息。这些文件此刻是孤儿——物理上存在，但没有任何 manifest 引用它们，所以任何查询都看不到。
commit 阶段：生成新的 metadata.json（包含新快照，新快照引用了这批文件），然后原子地把 Catalog 里的指针从旧版本切到新版本。
这个"原子地切指针"是整个机制的基石，它由 CAS（Compare-And-Swap，比较并交换）保证：提交时声明"我认为当前版本是 v37，请把它换成 v38"。如果此刻已经有别人把它改成了 v38，CAS 失败，本次提交需要重试。
为什么这一步能做到原子：因为它只修改一个指针，而不是修改一堆文件。切换前查询看到 v37 的全部文件，切换后看到 v38 的全部文件，不存在中间态。
故障发生在三个不同时点，分别怎么恢复
这是理解整套机制是否真的闭合的关键。
| 崩溃时点 | 此时的世界状态 | 恢复后发生什么 | 结果 |
|---|---|---|---|
| A. pre-commit 之前（正在写文件） | 部分 Parquet 文件在对象存储上，无人引用 | 从 checkpoint N-1 恢复，Kafka offset 回退，重新读、重新写新文件 | 正确。旧的孤儿文件成为垃圾，需要 remove_orphan_files 清理（见 §8） |
| B. pre-commit 之后、commit 之前 | 文件写好了，文件清单在 checkpoint state 里，但 Catalog 还没更新 | 从 checkpoint N（已完成）恢复，sink 从 state 里读出"待提交的文件清单"，重新执行提交 | 正确。这一步是整个机制能否闭合的命门——sink 必须把 pending 的事务信息持久化在 state 里，否则这批数据永远不会可见（丢数据） |
| C. commit 之后 | Catalog 已指向新快照 | 从 checkpoint N 恢复，sink 尝试再次提交同一批文件 | 正确，前提是提交操作幂等。Iceberg 的做法是记录已提交的 checkpoint 编号，重复提交会被识别并跳过 |
时点 B 值得再强调一次：很多"自己实现的 exactly-once sink"就死在这里——只做了"写完再提交"，但没有把 pending 事务持久化。一旦在 B 时点崩溃，恢复后 sink 不知道自己还欠一次提交，那批数据就永久丢失了，而且无声无息。
它在什么情况下不成立
这一节是本篇最实用的部分。宣称"我们做到了端到端 exactly-once"的系统，绝大多数在下面某一条上是破的。
情况一：Sink 既不支持事务也不支持幂等
2PC 要求参与者能做到"写了但不可见，随后可一次性生效"。如果 sink 是这些，它做不到：
- 调一个普通的 HTTP 接口（发出去就生效了，收不回来）
- 写一个没有唯一键约束的普通表（重复插入就是两行）
- 发一条短信、扣一次款（见情况六）
判断方法：问一句"这个写操作能不能撤销，或者重复执行两次结果和一次一样？"两个都答不出来，就不可能有 exactly-once。
情况二：外部系统的事务超时短于恢复时间
这是最隐蔽也最常踩的一条。
2PC 的参与者在 prepare 之后要一直持有未提交的事务。但外部系统通常给未提交事务设了超时——超时后自动回滚，防止资源被无限占用。
于是出现这个时间竞赛：
checkpoint N 完成(pre-commit) ──────────── 作业崩溃 ─────── 重启 ─────── 恢复并重新提交
│◀──────── 这段时间 ────────────────────▶│
必须 < 外部系统的事务超时
如果作业崩溃后重启花了 20 分钟（排队等资源、拉镜像、恢复大 state），而外部系统的事务 10 分钟就超时回滚了，那么恢复后的提交会失败——数据丢了。
以 Kafka sink 为例：Kafka 的事务型生产者有 transaction.timeout.ms 参数，broker 侧还有 transaction.max.timeout.ms 上限。这两个值与 checkpoint 间隔、以及最坏情况下的作业恢复时长之间存在硬约束。
情况三：下游读取时没有过滤未提交数据
即使 sink 端做对了，如果下游读的时候把"未提交"的数据也读了，一切白费。
Kafka 消费者有 isolation.level 参数：read_uncommitted 会读到事务中未提交的消息，read_committed 才只读已提交的。如果上游做了事务而下游用 read_uncommitted，端到端语义退化为 at-least-once。
Iceberg 这一侧天然没这个问题——未提交的文件不在任何 snapshot 里，读者根本看不见。这也是湖表格式相比"直接写文件"的一个实质优势。
情况四：源头本身就有重复
回看 12 篇：埋点数据在进入 Kafka 之前就已经带重复了（SDK 重试导致）。
Flink 的 exactly-once 保证的是"Kafka 里的每条消息，其影响恰好生效一次"。如果 Kafka 里同一个业务事件躺着两条消息，Flink 会忠实地把两条都精确处理一次，结果就是业务上重复。
情况五：作业以非正常方式重启
- 从更早的 checkpoint 恢复（比如手动指定回退到几小时前）：那之后已经 commit 的数据依然可见，重放会产生重复。这不是机制失效，是人为选择了"重放"。
- 无状态重启（丢弃 state 全新启动）：所有保证归零。
- 修改了作业拓扑后从 savepoint 恢复：如果算子的 UID 变了，对应 state 无法映射，那部分状态实际上是空的——这不会报错，作业能正常启动，只是结果错了。
情况六：副作用不可回滚
如果处理逻辑里包含"发短信""调用支付接口""推送通知"这类操作，2PC 完全无能为力——已经发出去的短信收不回来。
唯一的正确做法是把副作用移出流处理：流处理只负责把"应该发短信"这个事实写成数据（幂等地写入一张表），由一个独立的、自带幂等控制的下游服务去执行实际发送。
机制正确不代表工程可用。Flink 流式写 Iceberg 最主要的工程问题不是一致性，是小文件。
每次 checkpoint 完成时，sink 会关闭当前文件并提交。所以：
每次 commit 产生的文件数 ≈ sink 并行度 × 本批数据涉及的分区数
举一个具体的例子：
| 参数 | 值 |
|---|---|
| Sink 并行度 | 100 |
| 数据涉及的分区数（比如按小时 + 业务线分区） | 5 |
| checkpoint 间隔 | 1 分钟 |
| 一天的 checkpoint 次数 | 1440 |
一天产生的文件数 = 100 × 5 × 1440 = 720,000 个
每个文件的平均大小：假设日增数据 500 GB，则 500 GB ÷ 720,000 ≈ 0.7 MB/文件。
而 Parquet 的推荐文件大小通常在128 MB 到 1 GB 量级（这个范围来自社区通行经验，具体最优值取决于存储介质和查询模式）。0.7 MB 意味着比理想值小两到三个数量级。
很多人只知道"小文件不好"，但说不清机制。这里拆成两条完全独立的路径。
链路一：数据面开销（读文件本身）
文件数量多
↓
每个文件都要:发一次对象存储的 GET 请求(有固定往返延迟,通常几十毫秒)
+ 读 Parquet footer(拿 schema 和统计信息)
+ 解析 footer
↓
这套固定开销与文件大小无关 —— 读 0.7 MB 和读 700 MB 花的固定开销一样
↓
文件越小,固定开销占比越高
↓
72 万个文件 = 72 万次请求。即使并发 1000,也要 720 轮
关键点：对象存储的单次请求延迟远高于本地磁盘。存算分离架构把小文件问题放大了——这是"存算分离的代价"在这里的具体形态。
链路二：元数据面开销（查询规划阶段）
文件数量多
↓
Iceberg 的 manifest 里要为每个文件记一行(路径、行数、各列的 min/max 统计)
↓
manifest 文件本身变得巨大
↓
查询规划(planning)阶段要读完所有 manifest,才能知道该扫哪些文件
↓
规划时间从毫秒级涨到秒级甚至分钟级
↓
即使查询最终只需要读 3 个文件,也得先花 30 秒把 72 万条文件记录过一遍
链路二经常被忽略，但它更致命：因为它的开销发生在真正读数据之前，而且即使你的查询条件能过滤掉 99.99% 的文件，这个开销也躲不掉。谓词下推能减少要读的文件，但减少不了要检查的文件。
| 解法 | 做法 | 牺牲了什么 |
|---|---|---|
| ① 调大 commit 间隔 | checkpoint 从 1 分钟改成 10 分钟 | 数据可见延迟从 1 分钟变成 10 分钟。同时 checkpoint 间隔变大意味着故障恢复要重放更多数据 |
| ② 写入前重分布 | 在 sink 前加一次按分区键的 shuffle，让同一分区的数据集中到少数几个 sink 实例 | 引入了一次 shuffle（05 篇：shuffle 很贵）；而且如果分区数据倾斜（比如某个业务线占 80%），会造成 sink 实例负载不均，拖慢整个作业 |
| ③ 异步 compaction | 保持流式写入不变，另起一个批任务定期把小文件合并成大文件 | 额外的算力成本（要把数据读一遍再写一遍）；与流式写入争抢资源；存在提交冲突（见 后文）；合并期间存储要放大（新旧文件并存） |
实际做法通常是三者组合：commit 间隔取一个折中值（分钟级）、按分区做一次轻量重分布、再配一个低峰期跑的 compaction 任务。
-- Iceberg 的文件合并(以 Spark 存储过程的形式调用)
-- Iceberg 版本:0.14 及以后提供 rewrite_data_files 过程,具体参数名请以你的版本文档为准
CALL catalog.system.rewrite_data_files(
table => 'db.dwd_item_click_di',
strategy => 'binpack', -- binpack: 单纯按大小装箱合并,最省算力
options => map(
'target-file-size-bytes', '536870912', -- 目标文件大小 512 MB
'min-input-files', '10', -- 一个分区里至少有 10 个小文件才值得合并
'max-concurrent-file-group-rewrites', '5'-- 并发度上限,用来限制它抢多少资源
),
where => "dt = '2026-07-29'" -- 只处理指定分区,避免全表扫描
);
模拟输入：dt = '2026-07-29' 分区下有 30,000 个平均 0.7 MB 的文件，总计约 21 GB。
运行结果：
rewritten_data_files_count added_data_files_count rewritten_bytes
30000 42 22548578304
为什么是这个结果：21 GB 按 512 MB 目标大小装箱，理论上需要 42 个文件（21 GB ÷ 0.5 GB ≈ 42）。30,000 个输入文件被读取并重写为 42 个输出文件，文件数下降了约 714 倍—— 两条链路上的固定开销随之等比例下降。
compaction 是一个读全量、写全量的重任务。合并 21 GB 数据意味着 21 GB 的读 + 21 GB 的写，以及对应的 CPU 解压/压缩开销。
它会和这些东西抢资源：
- 流式写入作业：如果共用一个集群/队列，compaction 占满资源会导致 Flink 作业反压，进而 checkpoint 变慢甚至超时——而 checkpoint 超时会直接影响情况二里那个事务超时不等式
- 下游查询：compaction 期间对象存储的请求量和带宽被大量占用，查询延迟上升
- 对象存储的请求配额：云厂商对存储桶的请求速率通常有限制
通行做法：把 compaction 放在业务低峰期，并用独立的资源队列与线上作业隔离。
Iceberg 用 CAS 提交。现在有两个写入方同时在动同一张表：
时刻 T0: 表的当前版本 = v37
Flink sink: 读到 v37 ──▶ 写了新数据文件 ──▶ 尝试 CAS(v37 → v38) ──▶ ✅ 成功
compaction: 读到 v37 ──▶ 合并了旧文件 ──▶ 尝试 CAS(v37 → v38) ──▶ ❌ 失败(已经是 v38 了)
compaction 失败了怎么办？ Iceberg 会重试：重新读取当前版本 v38，检查自己合并的那些文件在 v38 里是否还存在且未被修改。
- 如果 Flink 只是追加了新文件，没动 compaction 正在合并的那些旧文件 → 冲突可以自动化解，compaction 基于 v38 重新提交为 v39
- 如果两者动了同一批文件（比如两个 compaction 任务撞了，或者 compaction 与行级更新撞了）→ 无法化解，其中一个必须放弃并重做
这就是乐观并发控制的固有代价：冲突不会导致数据错误（CAS 保证了这一点），但会导致做过的工作白费。冲突频率越高，浪费越多。
工程含义：compaction 任务的运行时间越长，它撞上冲突的概率越高，白干的代价也越大。 这就是为什么例子里要用 where 限定分区、用 max-concurrent-file-group-rewrites 限制并发——把一个大 compaction 拆成多个小的，单次冲突的损失就小。
Iceberg 的每次提交都产生一个新快照，旧快照及其引用的文件默认不会被删除——这正是 time travel 能成立的原因。
代价是存储成本没有上界。 一个每分钟提交一次的流式作业，一天产生 1440 个快照。加上 compaction 产生的新文件（旧文件仍被历史快照引用，删不掉），存储会持续膨胀。
两个必须配置的清理任务：
-- ① 过期旧快照:删除 7 天前的快照,以及仅被这些旧快照引用的数据文件
CALL catalog.system.expire_snapshots(
table => 'db.dwd_item_click_di',
older_than => TIMESTAMP '2026-07-23 00:00:00',
retain_last => 10 -- 无论如何至少保留最近 10 个快照,防止误删到当前可用版本
);
-- ② 清理孤儿文件:删除物理存在但不被任何快照引用的文件
-- 来源见 §4.4 时点 A —— 作业崩溃时写了一半的文件
CALL catalog.system.remove_orphan_files(
table => 'db.dwd_item_click_di',
older_than => TIMESTAMP '2026-07-29 00:00:00' -- 只清理 1 天前的,避免误删正在写入的文件
);
模拟输入：一张流式写入了 30 天的表，当前有 43,200 个快照，存储占用 18 TB。
运行结果（执行 ① 保留最近 7 天）：
deleted_data_files_count deleted_manifest_files_count
418,203 10,080
存储占用: 18 TB → 4.2 TB
为什么是这个结果：7 天之前的快照被删除后，那些仅被旧快照引用的数据文件失去了所有引用，可以物理删除。仍在最近 7 天快照中的文件保留。降幅取决于这段时间里 compaction 重写了多少数据——重写得越多，被替换掉的旧文件越多，可清理的空间越大。
remove_orphan_files的older_than参数必须谨慎设置。 如果设成"1 小时前"，而恰好有一个运行了 2 小时的写入任务正在写文件（还没提交，所以还是孤儿），这些文件会被当作垃圾删掉，导致该任务提交时找不到文件而失败。通行做法是设成至少 3 天，远大于任何单个写入任务的最长运行时间。
exactly-once 不是免费的，成本清单：
- checkpoint 必须成功且不能太慢，否则整个机制不成立
- sink 端要维护未提交事务，增加内存和状态开销
- 数据可见性延迟 = checkpoint 间隔（至少）
- §5 情况二那个不等式要一直成立，运维复杂度显著上升
下面这三种情况，用 at-least-once + 下游幂等更好：
① 下游天然幂等时。 如果最终结果是"按 key 覆盖写"（比如把用户画像写进 KV 存储，同一个 key 写两次结果一样），那么重复根本不产生影响。此时 at-least-once 就已经等价于 exactly-once，还省掉了 2PC 的全部开销。
② 结果本身是近似值时。 实时大屏的 UV、实时热榜——这类指标本来就带误差（12 篇 §5.3 说过实时链通常不做严格去重）。为 0.1% 的精度付 2PC 的代价不划算。
③ 延迟要求极高时。 exactly-once 的数据可见延迟至少等于一个 checkpoint 间隔。如果业务要求亚秒级可见（比如实时风控要在下单瞬间给结论），checkpoint 间隔不可能压到那么小，此时只能选 at-least-once 并在下游做幂等。
反过来，这些场景必须上 exactly-once：涉及金额的计费与结算、对外披露的经营数据、任何"多算一次就是资损"的场景。广告计费就是典型——后续会讲为什么广告计费链路的一致性要求是整条数据链路里最硬的。
小结
- exactly-once 不是"只传一次"，而是"重传重算多少次都行，但对最终可见状态的影响恰好等于一次"。实现的总纲是：让中间结果对外不可见，直到确认无误再一次性放出。
- 三段各自的保证拼不出端到端保证：Source 能重放 + 内部能恢复 = 一定会重算；重算 + Sink 立即可见 = 一定会重复。缺的是一个跨段协议。
- 2PC 的 prepare 阶段（做到只差最后一步但不可见）恰好提供了所需能力。Flink 的映射是：checkpoint = prepare，checkpoint 完成 = commit 决定，
notifyCheckpointComplete= commit 广播。 - Iceberg 侧靠 CAS 原子切换元数据指针实现"瞬间全部可见"。故障恢复的命门在 pre-commit 之后、commit 之前这个时点——sink 必须把待提交事务持久化在 state 里。
- 六种不成立的情况：sink 不支持事务或幂等 / 外部事务超时短于恢复时间 / 下游读未提交数据 / 源头本身有重复（exactly-once 的起点是 Kafka，不是用户手机） / 非正常重启 / 副作用不可回滚。
- 小文件从数据面（每文件固定的请求与 footer 开销）和元数据面（planning 要读完所有 manifest）两条独立路径拖慢查询，其中元数据面更致命，因为谓词下推躲不掉它。
- 三种解法各有牺牲：调大 commit 间隔牺牲延迟、写前重分布牺牲吞吐、异步 compaction 牺牲算力并引入冲突。乐观并发下的冲突不会导致数据错误，只会导致白干——所以 compaction 要切小。
- 快照和孤儿文件必须配定期清理，否则存储成本没有上界；
remove_orphan_files的时间阈值要远大于最长写入任务时长。 - 下游幂等、结果是近似值、或延迟要求极高时，at-least-once 是更好的选择。