Flink 端到端 Exactly-Once：两阶段提交实践

"数据不重不漏"是流处理里要求最高、也最容易做错的需求。Flink 端到端 Exactly-Once 的本质不是 Flink 单方面能做到的，而是三段能力的组合：数据源可以按位点回放（Kafka 天然支持）、Flink 通过 Checkpoint 保证状态只提交一次、输出端通过"事务"或"幂等"保证写外部系统不重复。写 Kafka 时，标准的做法是 Flink 的 Checkpoint + 两阶段提交（2PC）：Flink 在 Checkpoint 完成时统一提交 Kafka 事务，下游用 read_committed 的消费者读取，从而做到端到端精确一次。下面从原理到配置讲清楚怎么做、有哪些限制。

一致性语义与端到端挑战

先对齐三个概念：

• At-Most-Once：数据最多处理一次，可能丢。
• At-Least-Once：数据至少处理一次，故障重放时会重复。
• Exactly-Once：数据精确处理一次，不丢不重。

端到端一致性由"源 + 引擎 + Sink"三段共同决定。Flink 内部的状态一致性靠 Checkpoint 就能保证（故障后从最近一次成功的 Checkpoint 恢复，源侧按位点回放），但内部一致不代表输出一致：如果 Sink 在 Checkpoint 之后、外部系统确认之前失败，重放时同一批数据会被再写一次，落到下游就是重复。所以端到端 Exactly-Once 的难点在 Sink 侧——要么 Sink 支持事务（可回滚），要么写入本身幂等（按主键 upsert）。

Flink 内部 Checkpoint 与状态一致性

Flink 通过 Checkpoint 实现故障恢复：JobManager 周期性地向 Source 注入 barrier，barrier 随数据流传播，每个算子收到全部输入的 barrier 后把状态做快照；全部算子快照成功，Checkpoint 才算成功。Kafka Source 的 offset 提交与 Checkpoint 绑定：只有 Checkpoint 成功，消费位点才提交，恢复时从最近成功位点重放。这是内部 Exactly-Once 的基础，也是后续两阶段提交的"提交时机"来源。

两阶段提交原理

TwoPhaseCommitSinkFunction（Sink V1 时代的实现）把外部 Sink 包装成 2PC 参与者，生命周期如下：

• Pre-commit（预提交）：Checkpoint barrier 到达时，Sink 把所有待写数据写进一个事务并预提交，事务状态连同 Checkpoint 一起存进 Flink 状态。
• Commit（提交）：Checkpoint 成功后，Flink 通知 Sink 真正提交事务，数据对外可见。
• Abort（回滚）：Checkpoint 失败或作业恢复时，Flink 用状态里保存的事务句柄把未完成的事务回滚。

关键点：预提交状态保存在 Flink 状态里，所以作业重启后能找回"哪些事务待提交、哪些要回滚"，这正是"不重"的保证。在新版本中推荐使用 Sink V2 的 TwoPhaseCommittingSink 接口；旧的 TwoPhaseCommitSinkFunction 在 Flink 2.0 已被移除，迁移时要注意。

作业重启时的完整流程值得走一遍：假设 Checkpoint N 已经成功、对应事务 T1 已提交，Checkpoint N+1 进行到一半作业崩溃。恢复时 Flink 从 Checkpoint N 恢复状态，此时状态里保存的是"事务 T1 已提交、事务 T2 处于预提交"的元信息；T2 实际可能没有在 Kafka 侧完成预提交，Flink 会尝试对 T2 执行 abort，保证它不会在之后被提交。这个"预提交状态持久化 + 恢复时回滚未决事务"的机制，是 2PC Sink 崩溃后仍不重复的关键，也是事务状态必须随 Checkpoint 持久化、不能只放在内存里的原因。

Kafka 事务 Sink 实施与配置

Kafka 从 0.11 起支持事务，Flink 写 Kafka 的端到端 Exactly-Once 就是把"Kafka 生产者事务"和"Flink Checkpoint"绑定。核心配置：

• 生产者侧：设置 transactional.id，保证同一时间只有一个生产者实例在写；Flink Kafka Connector 通过 transactionalIdPrefix 生成，每个作业必须唯一。
• 语义设置：FlinkKafkaProducer 设 Semantic.EXACTLY_ONCE；SQL 建表时设 'sink.semantic' = 'exactly-once'，并配置 'sink.transactional-id-prefix'。
• 事务超时：transaction.timeout.ms 必须大于等于 Checkpoint 间隔与执行时长，同时小于等于 Broker 的 transaction.max.timeout.ms（默认 15 分钟），否则事务会被 Broker 提前终止。
• 消费侧：下游消费者必须设 isolation.level=read_committed，否则会读到未提交和回滚的数据。

一个 SQL 示例：

CREATE TABLE kafka_sink (
  user_id BIGINT,
  cnt BIGINT
) WITH (
  'connector' = 'kafka',
  'topic' = 'agg_result',
  'properties.bootstrap.servers' = 'broker:9092',
  'sink.semantic' = 'exactly-once',
  'sink.transactional-id-prefix' = 'agg-job-1',
  'properties.transaction.timeout.ms' = '300000'
);

同时把 Checkpoint 间隔设置得明显小于事务超时（比如 Checkpoint 1 分钟、事务超时 5 分钟），留足余量。

端到端 Exactly-Once 的限制与适用场景

必须清醒认识几条边界：

• 下游必须配合：只有 read_committed 的消费者才看不到未提交数据；任何 read_uncommitted 的消费方都会破坏"不重"的感知。
• 事务会产生空洞：被回滚的事务在消费端表现为一段"没有数据"的时间区间，对实时指标类消费可能造成瞬时空白。
• 有性能与延迟开销：每批事务的协调、日志写入都增加成本，Exactly-Once 通常比 At-Least-Once 延迟更高、吞吐更低，要按业务重要性取舍。
• 只适用于支持事务的系统：写 HBase、Redis、HTTP 服务这类没有 2PC 能力的 Sink，端到端 Exactly-Once 不成立，替代方案是幂等写入（唯一主键 + upsert）。
• Kafka 事务只保证单集群内：跨集群、跨系统（比如同时写 Kafka 和数据库）做不到单事务原子，需要接受最终一致性或引入外部协调。

幂等写入：没有事务时的替代方案

当 Sink 不支持事务时，端到端"不重不漏"退化为"At-Least-Once + 幂等去重"：允许重放，但重放不产生重复效果。实现依赖目标系统的唯一键能力：

• 写数据库：用主键 upsert，重复写入同一主键只生效一次。
• 写 Kafka：给消息加业务唯一键，下游按 Key 去重。
• 写文件系统：按 Checkpoint 边界命名文件，配合目标端"同名单文件覆盖"语义。

一个典型的幂等聚合示例（写入支持 upsert 的存储）：

INSERT INTO agg_result
SELECT user_id, count(*) AS cnt, sum(amount) AS amount
FROM ods_order
GROUP BY user_id;
-- agg_result 以 user_id 为主键，重复写入同一主键时按新值覆盖

这个方案没有 2PC 的事务开销和 read_committed 依赖，实现简单、吞吐高，但要求业务能接受"短暂重复读取后再收敛"的中间状态，对账口径上是最终一致而非实时一致。选择顺序建议：能接受最终一致 → 幂等写入；必须实时不重不漏 → 2PC 事务，且确认全链路（源、引擎、Sink、下游消费者）都支持。

常见坑与验证方法

高频踩坑清单：

• transactionalIdPrefix 在多个作业间重复，事务互相干扰。
• transaction.timeout.ms 小于 Checkpoint 时长，Checkpoint 还没提交事务就被 Broker 超时回滚。
• 下游消费者没设 read_committed，重复数据其实是读到了未提交/回滚内容。
• 多 Flink 作业写同一 topic 且各自开事务，事务协调器压力与 zombie producer 清理不及时。
• 作业重启后旧事务未 abort，出现"幽灵"积压或超时告警。
• Checkpoint 频繁失败时，事务反复 abort，吞吐骤降。
• Checkpoint 超时与事务超时互相矛盾（例如 execution.checkpointing.timeout 大于 transaction.timeout.ms），事务先于 Checkpoint 超时被回滚，作业进入"恢复-再失败"循环。

验证方法建议三步走：先做故障注入测试（作业运行中 kill 一个 TaskManager，重启后持续消费端抽样计数，确认无重复无丢失）；再用 read_committed 消费者对输出 topic 做总量对账（对比源 topic 的去重计数）；最后监控 Kafka 事务相关指标与端到端延迟，确认 Exactly-Once 的成本在可接受范围。注意故障注入要覆盖"Checkpoint 成功但提交前崩溃"和"提交后崩溃"两种时间点，只测一种容易得出错误的乐观结论；对账时用按业务 Key 去重后的计数，而不是简单看消息条数，因为事务回滚留下的空洞会让条数对不上。业务上如果只是"可容忍少量重复、不能丢"，At-Least-Once + 幂等去重往往比 2PC 更简单高效，不必为语义正确性付出全部性能代价。