采集到的数据有重复——Exactly-Once到底能不能保证？

重复数据的常见来源:
  1) 生产者重试:网络抖动时Producer.send重试,导致Kafka里有重复消息
     (Kafka 0.11+可以开idempotent producer避免,但仅限单会话)
 
  2) 消费者重启:Checkpoint/Offset未提交,重启后重新消费
 
  3) 作业升级:Savepoint恢复时回退几分钟数据
 
  4) 跨系统同步:A系统→Kafka→B系统,任何一环失败重试都产生重复
 
  5) 人工操作: 补数据/重跑调度
 
  结论: 重复是分布式系统的常态,必须靠幂等吸收

// Flink + Kafka 事务性Sink
FlinkKafkaProducer<String> kafkaProducer = new FlinkKafkaProducer<>(
    "output-topic",
    new SimpleStringSchema(),
    props,
    FlinkKafkaProducer.Semantic.EXACTLY_ONCE  // 核心: 事务语义
);
 
// 配合要求:
// 1) transaction.timeout.ms > flink checkpoint interval
props.setProperty("transaction.timeout.ms", "900000");  // 15分钟
 
// 2) Flink Checkpoint开启
env.enableCheckpointing(60_000);
env.getCheckpointConfig().setCheckpointingMode(CheckpointingMode.EXACTLY_ONCE);
 
// 3) 下游消费者必须配 isolation.level = read_committed
// 否则会读到未提交的事务数据
consumerProps.put("isolation.level", "read_committed");
 
// 工作原理: 2PC两阶段提交
// Phase-1: checkpoint barrier到达时,Sink开启Kafka事务写入
// Phase-2: checkpoint完成后,Sink commit事务 → 下游才能读到
// 如果checkpoint失败,事务abort → 数据不会被下游消费

-- ✓ 方案: 主键upsert保证幂等
-- 示例1: Delta Lake MERGE INTO
MERGE INTO dwd_orders t
USING stream_orders s ON t.order_id = s.order_id
WHEN MATCHED AND s.op_ts > t.op_ts
  THEN UPDATE SET *, t.op_ts = s.op_ts
WHEN NOT MATCHED
  THEN INSERT *;
-- 关键: 用op_ts版本号防止老数据覆盖新数据
 
-- 示例2: MySQL ON DUPLICATE KEY UPDATE
INSERT INTO dwd_orders (order_id, amount, status, op_ts)
VALUES (?, ?, ?, ?)
ON DUPLICATE KEY UPDATE
  amount = IF(VALUES(op_ts) > op_ts, VALUES(amount), amount),
  status = IF(VALUES(op_ts) > op_ts, VALUES(status), status),
  op_ts  = IF(VALUES(op_ts) > op_ts, VALUES(op_ts), op_ts);
 
-- 示例3: Hologres upsert(主键表)
INSERT INTO dwd_orders VALUES (...) ON CONFLICT(order_id) DO UPDATE SET ...;
 
-- 示例4: HBase/KV存储
-- put操作天然幂等(相同rowkey覆盖)

关键设计：① 必须有稳定主键；② 必须有版本号（op_ts或source_ts）防止乱序覆盖；③ Sink必须支持upsert原语（不是所有存储都支持，比如Hive就不行）。

方案三：去重机制（基于唯一ID）

当没有稳定主键或Sink不支持upsert时，靠"全局唯一event_id + 去重状态存储"做去重。这个方案通用性最强。

// Flink KeyedProcessFunction去重(基于State)
public class DedupFunction extends KeyedProcessFunction<String, Event, Event> {
    private transient ValueState<Boolean> seen;
    
    public void open(Configuration cfg) {
        StateTtlConfig ttl = StateTtlConfig
            .newBuilder(Time.hours(24))      // 24小时去重窗口
            .setUpdateType(StateTtlConfig.UpdateType.OnCreateAndWrite)
            .build();
        ValueStateDescriptor<Boolean> desc = new ValueStateDescriptor<>("seen", Boolean.class);
        desc.enableTimeToLive(ttl);
        seen = getRuntimeContext().getState(desc);
    }
    
    public void processElement(Event e, Context ctx, Collector<Event> out) throws Exception {
        if (seen.value() == null) {
            seen.update(true);
            out.collect(e);    // 第一次看到,输出
        }
        // 重复事件直接丢弃
    }
}
 
// 使用:
stream.keyBy(Event::getEventId)
      .process(new DedupFunction())
      .addSink(...);
 
// 海量数据场景:用Redis/RocksDB做外部去重存储
// 因为State有TTL,如果重复间隔超过TTL就会漏去重
// 金融场景建议TTL=7天,权衡存储成本和去重窗口

三方案对比

维度	端到端事务(2PC)	幂等写入	去重机制
严格程度	真Exactly-Once	等效Exactly-Once	等效(有TTL)
性能损失	20-30%	<5%	10-15%(State开销)
实现复杂度	高	低	中
依赖条件	所有组件支持事务	有主键+版本号	有唯一ID
适用场景	金融清结算	绝大多数业务	无主键数据
生产占比	约5%	约85%	约10%

面试高频追问

追问一：Kafka的幂等Producer就能保证Exactly-Once吗？

不能。Kafka的idempotent producer只保证"单个Producer会话内、单个partition上的去重"——同一次send的重试不会重复写。但跨会话（Producer重启）或跨partition就失效。严格Exactly-Once需要Kafka事务（transactional.id）+ 下游read_committed，这才能覆盖作业重启场景。面试时别混淆"idempotent"和"transactional"。

追问二：At-Least-Once + 幂等 跟 Exactly-Once 有什么实际差别？

结果上几乎没差别——最终状态都是一致的。差别在"中间状态"：Exactly-Once下游看不到未提交数据，At-Least-Once下游可能看到"被覆盖的中间版本"。对绝大多数业务无感（反正看最新结果），对风控/反欺诈这种要看"事件序列"的场景就关键了。生产上根据业务诉求选，不是越严格越好。

追问三：对方系统不支持upsert怎么做幂等？

三种退化方案：① 先delete再insert（要求有事务保证原子性，否则中间状态可能导致下游读到空）；② 写入时带unique约束，重复时忽略（INSERT IGNORE），缺点是更新会被忽略；③ 引入staging表+periodic swap，定期把staging数据merge到主表。最常见的还是方案三，牺牲实时性换幂等能力。

实战踩坑：Flink Sink用异步Flush导致重复

有个Flink作业sink写Hologres，为了性能开了async flush（攒10秒或1万条才真正写入），Checkpoint做完了但数据还在flush buffer里没落盘。作业挂掉重启后从checkpoint恢复，重新消费Kafka，buffer里的数据也写进去了——最终落库的是双份。

// ❌ 反模式: async Sink + Checkpoint不对齐
HologresSink sink = HologresSink.builder()
    .flushInterval(Duration.ofSeconds(10))  // 异步flush
    .flushBatchSize(10000)
    .build();
// Flink checkpoint完成时,buffer里的数据还没真正写入
// 作业挂重启 → checkpoint恢复消费 + buffer数据也写入 = 重复
 
// ✓ 正确做法: Checkpoint时强制flush
public class HologresCheckpointedSink extends RichSinkFunction<Row>
    implements CheckpointedFunction {
    
    public void snapshotState(FunctionSnapshotContext ctx) throws Exception {
        // Checkpoint barrier到达时,强制把buffer刷到Hologres
        // 只有flush完成,checkpoint才算成功
        flushBuffer();
        waitForAck();   // 等Hologres ack确认才返回
    }
    
    public void initializeState(FunctionInitializationContext ctx) { ... }
}
 
// 配合: 下游Hologres表用主键upsert,即使有重复写入也幂等
-- CREATE TABLE dwd_orders (order_id BIGINT PRIMARY KEY, ...);
-- 重复写入会被主键覆盖,结果幂等

解决方案：双保险——Sink在Checkpoint时强制flush + 下游主键幂等。这是生产上最稳的模式。单靠任何一个都不够：只flush没有幂等，丢数据的场景下还会漏；只幂等没有flush，checkpoint恢复时会覆盖更新状态。

总结

Exactly-Once在生产上几乎没有"纯技术"的方案，都是"At-Least-Once + 幂等/去重"的等效实现。严格2PC成本高，只在金融等极端场景用。生产最实用的是源端event_id + 下游主键upsert + op_ts版本号。面试时要讲清楚"真E2E vs 等效E2E的区别+为什么等效够用+具体实现"，比空喊口号分高得多。

下一篇聊云厂商产品的实战题：DataWorks数据集成同步速度慢，并发、批量、网络怎么调？

你们公司对一致性要求到什么级别？用的是哪种方案？评论区聊聊～

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