Kafka 事务 / EOS 在数仓链路里怎么用

难度与前置

难度：进阶
前置：知道消息可能重复；了解生产端 ack、消费位移提交；听说过 Exactly-Once。建议先读或对照 Checkpoint 与 Failover。
读完能做什么：说清 EOS 的适用范围与做不到的事，在数仓链路里选「Kafka 事务 / Flink EOS / 下游幂等」组合，并用重复注入验收。

适用场景

流式入仓 / 入湖：Flink ↔ Kafka ↔ 数仓表，要求故障恢复不丢不重（在声明边界内）
多 Sink（写 Kafka 又写外部系统）需要原子可见
面试与架构评审要讲清「正好一次」到底保什么

不适用：

纯离线 T+1 可重跑分区（优先幂等回刷）
下游无法幂等且不能开事务的黑盒 API（只能 at-least-once + 对账）
跨多个独立 Kafka 集群的「分布式事务幻想」

真实场景（一条）

点击流：Flink 消费 topic_click_raw，清洗后写 topic_click_dwd，再由另一作业入 Hive。曾用「先写 Kafka 再提交 checkpoint」，failover 后出现 1%～3% 重复点击，ADS UV 虚高。

改造：开启 Flink Kafka exactly-once sink（Transactional Idempotent Producer）+ Checkpoint 对齐；下游入仓按 event_id 幂等去重。故障演练注入重复后，DWD COUNT(DISTINCT event_id) 与 COUNT(*) 在去重表上一致，UV 偏差回到 < 0.2%。

怎么做（步骤清单）

1. 先对齐「EOS」三个层次

层次	保证什么	不保证什么
生产幂等	同 Producer Session 内单分区不重复写	跨会话、跨 Topic 原子
Kafka 事务	多分区写入 + 位移提交原子可见	事务外的 DB/文件
端到端 EOS（如 Flink）	Source 位移 + Sink 提交与 Checkpoint 对齐	任意外部系统未参与两阶段

数仓实践口诀：链路声明边界 + 关键出口幂等，不要迷信「开了 EOS 就永远不用对账」。

2. Kafka 事务生产端要点

properties

enable.idempotence=true
acks=all
transactional.id=ad-click-dwd-prod-1
max.in.flight.requests.per.connection=5

java

producer.initTransactions();
try {
  producer.beginTransaction();
  producer.send(record);
  // 与消费位移一并：sendOffsetsToTransaction(...)
  producer.commitTransaction();
} catch (Exception e) {
  producer.abortTransaction();
  throw e;
}

消费端读已提交：isolation.level=read_committed，否则可能读到事务中间态。

3. Flink 链路中的典型配置

java

FlinkKafkaProducer<String> sink = new FlinkKafkaProducer<>(
    topic,
    serializationSchema,
    props,
    FlinkKafkaProducer.Semantic.EXACTLY_ONCE
);
// 新版 Connector 使用 DeliveryGuarantee.EXACTLY_ONCE + transactionalIdPrefix

检查清单：

Checkpoint 开启且间隔合理（事务超时要覆盖对齐最坏情况）
transaction.timeout.ms（Kafka）> Checkpoint 时长预算

Checkpoint 间隔、对齐超时、failover 后状态恢复的运维细节以 Checkpoint 与 Failover 为准；本文只约定：Kafka 事务超时必须覆盖 Checkpoint 对齐最坏情况，两边手册交叉引用，避免各说各话。

transactionalIdPrefix 稳定且不同作业不冲突
并发与 Kafka 事务分区数匹配运维容量

4. 数仓落地：Sink 之外仍要幂等

即使 Kafka↔Flink EOS，写入 Hive/Iceberg/JDBC 仍可能：

作业重启产生「可见重复」若 Sink 非两阶段
下游批任务重跑叠加流式增量

5. 超时、僵尸事务与运维

事务超时过短 → 频繁 abort，吞吐下降
超时过长 → 读 read_committed 延迟变大
异常退出可能留下 pending 事务：用 Kafka 工具 / Broker 监控排查
作业水平扩缩容时评估 transactional id 与并发

bash

# 示意：查看消费者是否 read_committed、Lag 是否含未提交
kafka-consumer-groups.sh --bootstrap-server $BROKERS \
  --describe --group click_dwd_group

6. 选型决策树

text

能否接受重复 + 下游可去重？ → 常选 at-least-once（简单稳）
必须 Kafka 多分区原子可见？ → Kafka 事务 / Flink EOS
还要写外部 DB？ → DB 侧幂等或 Outbox，单靠 Kafka 事务不够
离线可重跑？ → 分区幂等优先于流式 EOS

常见坑（至少牢记这些）

以为 EOS 覆盖了 Hive 文件
未参与事务的文件系统写仍然可能重复。出口要幂等。
transaction.timeout < Checkpoint
大状态对齐时事务过期，陷入反复失败。
多作业共用 transactional.id 前缀
fencing 互踢，生产中断。
消费端仍用 read_uncommitted
读到后回滚的消息，制造「幽灵」与重复。
用 EOS 替代数据质量对账
时钟、上游补数、重复投递业务键仍会导致指标偏差。

如何验收（可量化）

[ ] 文档写清：保证范围（Topic A→B）与不保证范围（Hive/API）
[ ] Failover / Kill TM 演练：下游主键无新增重复（或重复率 < 阈值）
[ ] 注入重复 event_id：幂等表 COUNT(*)=COUNT(DISTINCT event_id)
[ ] 事务超时与 Checkpoint SLA 关系写进运维手册
[ ] isolation.level=read_committed 已在关键消费组生效

sql

-- 验收：分区内无重复业务键
SELECT dt, COUNT(*) AS cnt, COUNT(DISTINCT event_id) AS uv_key
FROM dwd_click_di
WHERE dt = '${biz_date}'
GROUP BY dt
HAVING COUNT(*) <> COUNT(DISTINCT event_id);

bash

# 演练记录
echo "kill_tm_at,dup_ratio_before,dup_ratio_after" >> eos_drill.csv

与 Checkpoint 的边界（互链）

问题	看哪篇
Checkpoint 间隔、对齐、failover 后从哪恢复	Checkpoint 与 Failover
Kafka 事务 / EOS 声明到哪、下游还要不要幂等	本篇
端到端延迟与乱序	实时链路延迟、乱序与重复
链路边界声明、注入演练与对账验收	端到端正好一次

两篇验收都要能答：Kill TM 之后，位移与 Sink 是否同一次 Checkpoint 提交；只开其一不能宣称端到端正好一次。

Kafka 事务 / EOS 在数仓链路里怎么用 ​

适用场景 ​

真实场景（一条） ​

怎么做（步骤清单） ​

1. 先对齐「EOS」三个层次 ​

2. Kafka 事务生产端要点 ​

3. Flink 链路中的典型配置 ​

4. 数仓落地：Sink 之外仍要幂等 ​

5. 超时、僵尸事务与运维 ​

6. 选型决策树 ​

常见坑（至少牢记这些） ​

如何验收（可量化） ​

与 Checkpoint 的边界（互链） ​

相关面试题 ​