Flink作业挂了重启后数据重复了——Exactly-Once到底怎么保证？

数仓项目手册第 15 题约 11 分钟读完

本文目录

先说结论
Checkpoint：Flink内部的Exactly-Once
方案一：两阶段提交（2PC）——写Kafka Sink
方案二：幂等写入——写OLAP/数据库
两种方案对比
生产环境的完整方案：三层保障
面试高频追问
实战踩坑
总结

先说结论

Exactly-Once有三个层次：① Source端（Kafka Consumer）→ Flink内部 → ② Sink端（写入下游）。Checkpoint只保证第一层和中间层的Exactly-Once，第二层需要额外机制。

端到端Exactly-Once的两种方案：① 两阶段提交（2PC）——Flink原生支持Kafka Sink的2PC，写Kafka天然Exactly-Once；② 幂等写入——下游支持Upsert（如Doris/Hologres/Redis），重复写入不会产生重复数据。

实际生产中，幂等写入比2PC更实用。因为2PC只有Kafka Sink支持得好，写OLAP引擎、MySQL、HBase都不支持2PC，只能靠幂等写入。

先把这段讲顺；面试官追问时，再展开下面的原理和细节。

详细讲解

实时数仓最怕的事情之一就是'数据重复'。Flink作业因为OOM或者Checkpoint超时重启了，重启后从上一次成功的Checkpoint恢复——但中间那段已经写入下游的数据又会被重新处理和写入，导致重复。面试官问'你怎么保证Exactly-Once'，很多人只能说'开Checkpoint就行了'——但Checkpoint只保证Flink内部的Exactly-Once，端到端的Exactly-Once还需要下游配合。我在某元宇宙社交应用从0搭建Flink实时数仓时，在这个问题上踩了很多坑，今天把完整的方案讲清楚。

Checkpoint：Flink内部的Exactly-Once

先理解Checkpoint的机制——这是Flink内部Exactly-Once的基础：

Checkpoint机制简化流程：
 
JobManager定期发起Checkpoint：
  1. 向Source发送Barrier（检查点屏障）
  2. Barrier随数据流向下游传递
  3. 每个算子收到Barrier时：
     - 暂停处理（Barrier对齐）
     - 把当前State快照写入持久化存储（HDFS/S3）
     - 继续处理
  4. 所有算子都完成快照 -> Checkpoint成功
 
作业重启时：
  1. 从最近一次成功的Checkpoint恢复State
  2. Kafka Consumer从Checkpoint记录的offset重新消费
  3. 重新处理Checkpoint之后的数据
  -> Flink内部不会丢数据也不会重复处理

关键问题在这里：Checkpoint保证了Flink内部的Exactly-Once——从Kafka读取和Flink内部计算不会重复。但Checkpoint完成后、下一次Checkpoint之前，Flink可能已经把结果写入了下游。作业重启后，这部分数据会被重新处理和重新写入——导致下游数据重复。

方案一：两阶段提交（2PC）——写Kafka Sink

如果Flink的输出目标是Kafka（比如实时DWD写回Kafka Topic），可以用Flink原生的两阶段提交实现端到端Exactly-Once：

// Flink写Kafka开启Exactly-Once
Properties kafkaProps = new Properties();
kafkaProps.setProperty("bootstrap.servers", "kafka:9092");
// 开启Exactly-Once的关键配置
kafkaProps.setProperty("transaction.timeout.ms", "900000");
 
FlinkKafkaProducer<String> sink = new FlinkKafkaProducer<>(
    "dwd_user_event",              // 目标Topic
    new SimpleStringSchema(),
    kafkaProps,
    FlinkKafkaProducer.Semantic.EXACTLY_ONCE  // 关键设置
);
 
// 原理：
// 1. Checkpoint开始时：Kafka Producer开启事务
// 2. 正常处理期间：数据写入Kafka但未提交（预提交）
// 3. Checkpoint完成时：提交Kafka事务（正式提交）
// 如果Checkpoint失败：事务回滚，已写入的数据不可见

注意事项：① Kafka的transaction.timeout.ms必须大于Checkpoint间隔，否则事务可能超时被Kafka自动回滚；② 下游消费者必须设置isolation.level=read_committed，否则会读到未提交的数据；③ 2PC会增加端到端延迟（数据在Checkpoint完成前对下游不可见）。

方案二：幂等写入——写OLAP/数据库

大多数实时数仓的最终目标不是Kafka，而是OLAP引擎（Doris/Hologres/ClickHouse）或数据库。这些系统不支持2PC，必须用幂等写入来保证不重复：

-- 幂等写入方案：Upsert模式（Doris/Hologres）
-- 相同主键的数据会覆盖更新，不会重复插入
 
-- Doris建表（Unique Key模型）
CREATE TABLE dws_gift_summary_realtime (
    window_start   DATETIME,
    gift_id        VARCHAR(64),
    gift_cnt       BIGINT,
    send_uv        BIGINT,
    total_coins    BIGINT
) UNIQUE KEY(window_start, gift_id)  -- 主键：窗口+礼物ID
DISTRIBUTED BY HASH(gift_id) BUCKETS 16;
 
-- Flink写入时：相同window_start+gift_id的数据覆盖更新
-- 即使Flink重启重放了数据，写入的结果也是一样的
-- 因为UNIQUE KEY模型保证了：相同主键只保留最新一条

两种方案对比

维度	两阶段提交（2PC）	幂等写入
原理	Checkpoint时提交事务	重复写入不产生重复数据
支持的Sink	Kafka（原生支持）	Doris/Hologres/MySQL/Redis等
延迟影响	有（数据等Checkpoint完成才可见）	无（数据立即可见）
实现复杂度	配置即可（Flink原生）	需要设计主键和Upsert策略
前提条件	Kafka事务支持	下游支持Upsert/Replace
实际使用频率	中（只用在Kafka Sink）	高（大多数OLAP场景）

生产环境的完整方案：三层保障

在生产环境中，我们用三层保障来确保端到端数据不重复：

三层保障体系：
 
第一层：Flink Checkpoint（保证Flink内部Exactly-Once）
  - Checkpoint间隔：60秒
  - Checkpoint超时：120秒
  - State Backend：RocksDB（大状态）
  - 保留最近3个Checkpoint
 
第二层：幂等写入（保证下游不重复）
  - Doris/Hologres：Unique Key模型 + Upsert写入
  - Redis：SET命令天然幂等
  - Kafka：EXACTLY_ONCE语义
 
第三层：离线修正（最终一致性兜底）
  - 每天凌晨离线链路重算，覆盖实时数据
  - 即使实时有微小误差，T+1一定正确

面试高频追问

追问一：Checkpoint间隔设多少合适？

这是一个权衡问题。间隔太短（比如5秒）：Checkpoint太频繁，State快照的IO开销大，影响正常处理性能。间隔太长（比如10分钟）：作业重启后要重放10分钟的数据，恢复时间长，而且2PC场景下数据可见延迟高达10分钟。我们的经验是：大多数场景设60秒——State快照的开销可接受，重启恢复也只需重放1分钟数据。State特别大的作业（GB级别）可以适当调长到120-180秒。

追问二：Checkpoint一直失败怎么排查？

Checkpoint失败是Flink最常见的生产问题，常见原因有三个：① Barrier对齐超时——某个算子处理速度跟不上（通常是反压导致），Barrier等太久就超时了。解决方案是开启Unaligned Checkpoint（Flink 1.11+），不需要等Barrier对齐。② State太大——State快照写HDFS/S3太慢，超过Checkpoint超时时间。解决方案是增大超时时间，或者优化State大小（及时清理过期State）。③ 外部依赖超时——Checkpoint过程中需要跟外部系统交互（如2PC的Kafka事务），网络抖动导致超时。解决方案是增大事务超时时间。

追问三：At-Least-Once和Exactly-Once怎么选？

如果下游是幂等的（Doris Unique Key、Redis SET），其实用At-Least-Once + 幂等写入就够了——性能更好（不需要Barrier对齐），效果跟Exactly-Once一样。只有在下游不是幂等的场景（比如写Kafka给下游消费、写MySQL的非主键表、做累加计数器），才必须用Exactly-Once。我们的实际做法是：Flink内部开Exactly-Once（保证State一致性），Sink端根据下游类型选择——写Doris用At-Least-Once + Upsert，写Kafka用Exactly-Once + 2PC。

实战踩坑

我在某元宇宙社交应用遇到过一次经典的数据重复事故。当时Flink作业写入Doris，用的是Doris的Duplicate Key模型（允许重复插入）而不是Unique Key模型。Flink作业OOM重启后，从Checkpoint恢复重放了20分钟的数据，全部重复写入Doris——实时大屏上那20分钟的送礼金额直接翻倍了。

紧急修复：① 手动删除Doris中重复时间段的数据，用离线数据覆盖写入；② 长期方案是把Doris的所有实时写入表改成Unique Key模型，主键设为window_start + 业务维度键。改完后即使Flink重复写入，Doris也会自动去重。

-- 踩坑：Doris表模型选错导致重复数据
 
-- Duplicate Key模型（允许重复！）
CREATE TABLE dws_gift_summary_rt (
    window_start DATETIME,
    gift_id VARCHAR(64),
    gift_cnt BIGINT
) DUPLICATE KEY(window_start, gift_id)  -- 不去重！
DISTRIBUTED BY HASH(gift_id) BUCKETS 16;
-- Flink重启重放后：同一window有两行
 
-- Unique Key模型（自动去重）
CREATE TABLE dws_gift_summary_rt (
    window_start DATETIME,
    gift_id VARCHAR(64),
    gift_cnt BIGINT
) UNIQUE KEY(window_start, gift_id)     -- 相同主键只保留最新！
DISTRIBUTED BY HASH(gift_id) BUCKETS 16;
-- Flink重启重放后：同一主键覆盖更新，不会重复

教训：实时写入OLAP引擎时，一定要用Unique Key或Aggregate Key模型，千万不要用Duplicate Key。Duplicate Key适合日志分析（不需要去重），但实时数仓必须要去重能力。这一点在Doris、StarRocks里都适用。

总结

端到端Exactly-Once不是'开一个配置就搞定了'的事情——需要Source端（Checkpoint恢复offset）、Flink内部（State一致性）、Sink端（2PC或幂等写入）三层配合。实际生产中，幂等写入是最实用的方案，适用于绝大多数OLAP场景。面试时讲Exactly-Once，不要只说'开Checkpoint'——要能说清楚三个层次各自的机制、幂等写入怎么设计主键、以及你遇到过的重复数据问题和解决方案。

Exactly-Once搞定了，下一篇聊一个贴近业务的问题——实时大屏要秒级延迟，但Flink窗口是1分钟，怎么做到'看起来是实时'的效果？

你们的Flink作业Checkpoint间隔设了多少？有没有遇到过数据重复的问题？评论区聊聊～

关注「Wood聊大数据」| 9年大数据老兵，聊技术、聊职场、聊成长

数仓项目实战面试通关手册

第16-20篇：实战亮点与面试收官

Wood聊大数据

9年大数据老兵，聊技术、聊职场、聊成长

刷题进度保存在当前浏览器