4、事务性写入（Exactly-Once）：Exactly-Once语义详解、两阶段提交（2PC）在OLAP中的应用、Kafka与OLAP引擎的Exactly-Once集成

聊到数据一致性，最硬核的话题就是 Exactly-Once 语义。说白了，就是每条数据不多不少，恰好被处理一次。在 OLAP 场景里，这几乎是终极目标。

我刚开始接触这个领域时，觉得这玩意儿就是个理论概念。后来在做一个实时数仓项目，数据总是对不上账，排查了三天才发现是 Kafka 重复消费导致的。嗯，从那以后，我再也不敢小看 Exactly-Once 了。

4.1 Exactly-Once 语义详解

先理清几个基础概念。消息投递语义通常分三种：

语义级别	含义	典型场景
At-Most-Once	最多一次，丢了不重试	日志采集，允许丢少量数据
At-Least-Once	至少一次，可能重复	大部分消息队列默认模式
Exactly-Once	恰好一次，不丢不重	金融、对账、实时报表

你想想看，为什么 Exactly-Once 这么难？因为分布式系统里，网络可能超时、节点可能宕机。一个写入请求发出去，如果没收到响应，你是重试还是不重试？重试可能重复，不重试可能丢数据。

我个人习惯把 Exactly-Once 拆成两个层面来理解：

• 端到端 Exactly-Once：从数据源到最终存储，整个链路保证不丢不重。
• 存储层 Exactly-Once：只保证写入引擎时的幂等性，上游重复数据由引擎去重。

在 OLAP 场景里，我们通常追求的是 存储层 Exactly-Once。因为端到端太难了，涉及多个系统协调，代价极高。

4.2 两阶段提交（2PC）在 OLAP 中的应用

两阶段提交（2PC）是实现 Exactly-Once 的经典方案。我在项目中用过好几次，每次都要小心翼翼。

2PC 分为两个阶段：

1. 准备阶段（Prepare）：协调者问所有参与者，能不能提交？参与者写日志、锁资源，然后回复 Yes/No。
2. 提交阶段（Commit/Abort）：如果所有人都说 Yes，协调者发 Commit；否则发 Abort。

听起来简单，但坑不少。我曾经在一个 ClickHouse 集群上做 2PC 写入，结果协调者挂了，所有参与者都处于「预提交」状态，资源锁了半小时。那叫一个酸爽。

在 OLAP 引擎里，2PC 主要用于 跨分片写入。比如一张表有 10 个分片，写入时要么全部成功，要么全部回滚。下面是一个简化的事务写入流程：

-- 伪代码：OLAP 引擎的 2PC 写入
BEGIN TRANSACTION;

-- Phase 1: Prepare
FOR each shard IN shards:
    shard.prepare(write_batch);
    IF shard.status == FAILED:
        ABORT TRANSACTION;

-- Phase 2: Commit
FOR each shard IN shards:
    shard.commit(write_batch);

END TRANSACTION;

这里要注意，OLAP 引擎的 2PC 和传统数据库的 2PC 有个关键区别：OLAP 的写入通常是 批量 的，而不是单条。所以 Prepare 阶段锁的是整个批次，而不是单行。这能减少锁冲突，但也会放大故障影响范围。

4.3 Kafka 与 OLAP 引擎的 Exactly-Once 集成

Kafka 从 0.11 版本开始支持 Exactly-Once 语义，核心是 事务性生产者 和 幂等性消费者。我最近在做一个实时数仓项目，就是拿 Kafka 对接 ClickHouse，实现端到端的 Exactly-Once。

整体架构是这样的：

具体实现时，有几个关键步骤：

4.3.1 Kafka 事务性生产者配置

// Java 示例：Kafka 事务性生产者
Properties props = new Properties();
props.put("bootstrap.servers", "localhost:9092");
props.put("transactional.id", "my-transactional-id");  // 必须设置
props.put("enable.idempotence", "true");               // 开启幂等

KafkaProducer<String, String> producer = new KafkaProducer<>(props);
producer.initTransactions();

try {
    producer.beginTransaction();
    producer.send(new ProducerRecord<>("topic", "key", "value"));
    producer.send(new ProducerRecord<>("topic", "key2", "value2"));
    producer.commitTransaction();
} catch (Exception e) {
    producer.abortTransaction();
}

这里有个细节：transactional.id 必须全局唯一。如果两个 Producer 用了同一个 ID，Kafka 会认为前一个挂了，强制中止它的未完成事务。我遇到过因为配置错误导致事务冲突，数据写不进去的情况。

4.3.2 OLAP 引擎的幂等写入

光靠 Kafka 的事务还不够，OLAP 引擎本身也要支持幂等。以 ClickHouse 为例，我常用的方案是 ReplacingMergeTree + 版本号：

-- ClickHouse 表定义：支持幂等写入
CREATE TABLE my_table (
    id UInt64,
    event_time DateTime,
    data String,
    _version UInt64  -- 业务版本号，用于去重
) ENGINE = ReplacingMergeTree(_version)
ORDER BY (id, event_time);

写入时，如果遇到重复的 id + event_time，引擎会根据 _version 保留最新的那条。这样即使 Consumer 重复消费了 Kafka 消息，最终数据也是正确的。

4.3.3 两阶段提交在 Kafka-OLAP 集成中的变体

有些场景下，我们需要在 Kafka 和 OLAP 引擎之间做 跨系统事务。比如：从 Kafka 消费一批数据，写入 ClickHouse，然后提交 offset。这三个操作要么全成功，要么全失败。

这时候可以用 Kafka 事务 + OLAP 引擎的 2PC 组合。流程如下：

1. Consumer 开启 Kafka 事务，从 Kafka 拉取消息。
2. Consumer 调用 OLAP 引擎的 Prepare 接口，写入预提交数据。
3. 如果 Prepare 成功，Consumer 提交 Kafka 事务，然后调用 OLAP 引擎的 Commit。
4. 如果任何一步失败，回滚 Kafka 事务和 OLAP 引擎的预提交。

这个方案看起来很完美，但实际落地时要注意性能。2PC 的 Prepare 阶段会锁住 OLAP 引擎的分片，高并发下容易死锁。我建议只在 小批量、低延迟 的场景下使用，比如实时风控、交易对账。