3、Flink端到端Exactly Once实现：两阶段提交协议、Kafka Sink的Exactly Once、JDBC Sink的Exactly Once

好，咱们进入第三章。这一章是实时ETL链路里最硬核的部分——端到端的Exactly Once。说白了，就是保证数据从源头到终点，不丢不重，只处理一次。

很多朋友问我：“Flink不是有Checkpoint吗？为什么还要搞什么两阶段提交？” 嗯，这里要注意——Checkpoint保证的是Flink内部状态的一致性，但数据一旦写到外部系统（比如Kafka、MySQL），Flink就管不着了。外部系统需要自己配合Flink，才能实现端到端的Exactly Once。

我个人习惯把这个问题拆成三层来看：状态后端（内部）、Source（源头）、Sink（终点）。这一章我们重点讲Sink这一层，也就是数据怎么写到外部系统，还能保证不重复。

3.1 两阶段提交协议（2PC）

两阶段提交协议，简称2PC。它不是一个新概念，分布式数据库里早就用了。Flink把它搬过来，作为实现Sink端Exactly Once的基石。

2PC把一次事务提交分成两个阶段：

• 预提交阶段（Pre-commit）： 所有参与者准备好提交，并告诉协调者“我准备好了”。
• 提交阶段（Commit）： 协调者收到所有参与者的“准备好”信号后，广播“正式提交”。

Flink是怎么做的呢？它把Checkpoint的Barrier当作2PC的“协调者”。当Barrier流过算子时，Sink会先执行预提交，把数据写入外部系统的临时事务中。等Checkpoint完成，再触发正式提交。

我画了一张图，帮你理解这个流程：

你想想看，这个机制保证了什么？如果任务在预提交之后、正式提交之前挂了，Flink重启后会根据Checkpoint的状态，回滚所有未提交的事务。数据不会丢失，也不会重复。

3.2 Kafka Sink的Exactly Once

Kafka Sink是Flink里用得最多的Sink之一。Kafka从0.11版本开始支持事务，这为Exactly Once提供了基础。

Flink的Kafka Sink实现Exactly Once，依赖两个核心机制：

• Kafka事务API： 用于在Kafka内部开启、提交、回滚事务。
• Flink的TwoPhaseCommitSinkFunction： 这是Flink提供的抽象类，封装了2PC的逻辑。

配置起来其实不复杂。我直接给你看代码：

// 开启Kafka Sink的Exactly Once语义
DataStream<String> stream = ...;

stream.addSink(
    FlinkKafkaProducer.<String>builder()
        .setBootstrapServers("localhost:9092")
        .setTopic("output-topic")
        .setRecordSerializer(
            new KafkaRecordSerializationSchema<>() {
                @Override
                public ProducerRecord<byte[], byte[]> serialize(
                        String element, 
                        Long timestamp, 
                        SerializationSchema.TargetWrapper target) {
                    return new ProducerRecord<>(
                        "output-topic", 
                        element.getBytes(StandardCharsets.UTF_8));
                }
            })
        // 关键配置：设置语义为EXACTLY_ONCE
        .setSemantic(Semantic.EXACTLY_ONCE)
        // 事务超时时间，默认1小时
        .setTransactionalIdPrefix("my-etl-")
        .build()
);

这里有个坑，我必须提醒你：事务超时时间。Kafka的事务默认超时是15分钟，但Flink的Checkpoint间隔可能更长。如果Checkpoint耗时超过15分钟，Kafka会自动中止事务，导致数据丢失。

Kafka Sink的Exactly Once还有一个前提：消费者必须设置隔离级别为read_committed。否则消费者会读到未提交的数据，造成数据不一致。

// 消费者端配置
Properties props = new Properties();
props.put("isolation.level", "read_committed");

3.3 JDBC Sink的Exactly Once

JDBC Sink的Exactly Once实现起来比Kafka麻烦一些。因为大多数关系型数据库（MySQL、PostgreSQL等）不支持分布式事务，或者性能开销太大。

Flink官方没有提供JDBC的Exactly Once Sink，但社区和第三方库（如Apache Bahir）提供了实现。核心思路是：利用数据库的幂等性 + 主键去重。

具体做法有两种：

方案	原理	优点	缺点
幂等写入	使用 INSERT ... ON DUPLICATE KEY UPDATE 或 MERGE 语句	实现简单，不需要事务	需要表有唯一键；重复写入有性能开销
事务写入	利用数据库单机事务 + Flink Checkpoint	严格Exactly Once	事务粒度粗，吞吐量受限

我个人更推荐幂等写入方案。为什么？因为它在生产环境里更稳定，不容易出现事务死锁。你想想看，实时ETL的吞吐量动辄每秒几万条，如果每条数据都开一个数据库事务，数据库根本扛不住。

下面是一个基于幂等写入的JDBC Sink示例：

// 使用JDBC幂等写入实现Exactly Once
DataStream<Event> stream = ...;

stream.addSink(
    JdbcSink.sink(
        // SQL语句：如果主键冲突则更新
        "INSERT INTO events (id, name, ts) VALUES (?, ?, ?) " +
        "ON DUPLICATE KEY UPDATE name = VALUES(name), ts = VALUES(ts)",
        (statement, event) -> {
            statement.setString(1, event.getId());
            statement.setString(2, event.getName());
            statement.setLong(3, event.getTimestamp());
        },
        JdbcExecutionOptions.builder()
            .withBatchSize(1000)
            .withBatchIntervalMs(200)
            .build(),
        new JdbcConnectionOptions.JdbcConnectionOptionsBuilder()
            .withUrl("jdbc:mysql://localhost:3306/mydb")
            .withDriverName("com.mysql.cj.jdbc.Driver")
            .withUsername("root")
            .withPassword("password")
            .build()
    )
);

还有一种场景——你要写入的数据库不支持 ON DUPLICATE KEY UPDATE（比如某些老版本的PostgreSQL）。这时候可以用事务方案：在Checkpoint的预提交阶段，把数据写入临时表；正式提交时，把临时表的数据移到目标表。

嗯，这里要注意：事务方案虽然严格，但实现起来复杂，而且容易造成数据库锁竞争。我建议只在数据一致性要求极高、且吞吐量不大的场景下使用。

3.4 两种Sink的对比与选型

我把Kafka Sink和JDBC Sink的Exactly Once实现做个对比，方便你选型：

维度	Kafka Sink	JDBC Sink
实现方式	两阶段提交（2PC）	幂等写入 / 单机事务
依赖	Kafka事务API	数据库唯一约束 / 事务支持
吞吐量	高（批量提交）	中（受限于数据库写入能力）
实现复杂度	低（Flink内置支持）	中（需要自己写SQL）
适用场景	高吞吐、低延迟的流式写入	低吞吐、强一致性的批量写入

说白了，如果你的下游是Kafka，直接用Flink内置的Exactly Once支持就行，省心省力。如果下游是数据库，优先考虑幂等写入方案，实在不行再上事务方案。