3. 幂等写入机制：从理论到落地的完整方案

幂等性，这个词听起来有点学术。说白了就是：同一个操作，无论执行多少次，结果都一样。在OLAP场景里，这太重要了。你想想看，数据管道偶尔重跑、任务失败重试、消息队列重复投递——这些情况天天发生。如果没有幂等机制，数据就会翻倍、乱掉、对不上。

我个人习惯把幂等写入比作「盖章」。你盖一次是那个红印，盖十次还是那个红印，不会变成十个不同的印。嗯，这个比喻虽然简单，但核心思想全在里面了。

3.1 幂等性的核心概念

先搞清楚几个关键点。幂等性在OLAP写入中，主要解决的是重复数据问题。我遇到过不止一次，凌晨三点ETL任务挂了，运维重启后没检查偏移量，结果同一批数据写了两遍。第二天对账，数据对不上，排查到天亮。

幂等写入的三个层次：

层次	含义	典型场景
请求幂等	同一条写入请求重复发送，结果一致	API重试、消息队列重复消费
批次幂等	同一个数据批次重复处理，结果一致	Spark/Flink任务失败重跑
事务幂等	跨多个表的写入操作，整体幂等	数据湖多分区写入、CDC同步

这里有个容易混淆的地方：幂等不等于去重。去重是发现重复后删除，幂等是从设计上保证不会产生重复。前者是事后补救，后者是事前预防。我个人更倾向于幂等设计，因为去重本身也有开销和风险。

3.2 基于唯一键的去重策略

唯一键是幂等写入的基石。没有唯一键，你根本没法判断两条数据是不是同一个东西。我见过很多团队在数据量小的时候不设唯一键，等数据涨到几亿条才发现重复率高达5%，那时候再想清洗，成本就大了。

唯一键的选择有讲究：

• 业务主键：比如订单ID、用户ID。最自然，但要注意业务变更（比如订单ID生成规则改了）。
• 复合唯一键：多个字段组合。比如日期+店铺ID+商品ID。适合明细数据。
• 消息唯一ID：由上游系统生成，比如Kafka消息的offset+partition。适合流式写入。
• 哈希指纹：对整行数据做MD5或SHA256。适合无业务主键的场景，但要注意哈希碰撞。

我曾经在一个实时数仓项目里踩过坑。当时用了业务主键做去重，结果业务方改了订单ID的生成规则，导致新旧ID冲突。从那以后，我养成了一个习惯：唯一键必须包含时间戳或版本号，至少能区分不同时期的数据。

3.3 实现幂等写入的三种常见方案

方案不在多，在于适用。我这些年用下来，真正能打的就是下面三种。每种都有它的适用场景和坑，我一个个说。

方案一：基于数据库的唯一约束

这是最直接的办法。在目标表上建立唯一索引或主键约束，写入时使用 INSERT ... ON DUPLICATE KEY UPDATE（MySQL）或 MERGE INTO（ClickHouse）或 UPSERT（Doris）。

举个例子，在ClickHouse里：

-- 创建ReplacingMergeTree表，用order by定义唯一键
CREATE TABLE orders_uid (
    order_id String,
    user_id UInt32,
    amount Decimal(10,2),
    event_time DateTime
) ENGINE = ReplacingMergeTree(event_time)
ORDER BY (order_id, user_id);

-- 写入时，相同order_id+user_id的数据会被去重
INSERT INTO orders_uid VALUES 
('ORD001', 1001, 99.00, '2024-01-01 10:00:00'),
('ORD001', 1001, 99.00, '2024-01-01 10:00:00');  -- 重复写入，最终只保留一条

这个方案的优点是简单、可靠。缺点是：

• 对数据库引擎有要求，不是所有OLAP引擎都支持真正的UPSERT
• 唯一约束在高并发下可能成为瓶颈
• 部分引擎（如Hive）不支持行级更新

方案二：基于外部存储的幂等校验

当数据库本身不支持幂等写入时，可以用外部存储来记录已处理的数据ID。常见做法是用Redis或HBase维护一个「已处理ID集合」。

流程是这样的：

1. 每条数据到来时，先查Redis里有没有这个唯一ID
2. 如果有，直接跳过（已处理过）
3. 如果没有，先写入Redis（标记为已处理），再写入目标表
4. 写入成功后，更新Redis中的状态（可选）

代码示例（伪代码）：

def write_with_idempotency(data, unique_key):
    # 1. 检查是否已处理
    if redis_client.exists(f"processed:{unique_key}"):
        logger.info(f"数据已处理，跳过: {unique_key}")
        return
    
    # 2. 标记为处理中（防止并发重复）
    redis_client.set(f"processing:{unique_key}", "1", ex=300)
    
    try:
        # 3. 写入目标表
        database.insert(data)
        
        # 4. 标记为已处理
        redis_client.set(f"processed:{unique_key}", "1", ex=86400)
        redis_client.delete(f"processing:{unique_key}")
    except Exception as e:
        # 5. 失败时清理标记，允许重试
        redis_client.delete(f"processing:{unique_key}")
        raise e

这个方案的好处是通用性强，任何存储引擎都能用。但要注意几个坑：

• Redis挂了怎么办？——需要做高可用，或者降级为数据库去重
• Redis内存不够怎么办？——设置过期时间，或者用布隆过滤器减少内存占用
• 并发写入时两个线程同时查到「未处理」怎么办？——用Redis的SETNX或Lua脚本做原子操作

我记得有一次，Redis集群因为网络分区导致部分key丢失，结果数据重复写入了。从那以后，我加了一层「双检」逻辑：Redis查到已处理，还要去数据库确认一下。虽然多了一次查询，但安全多了。

方案三：基于分区覆盖的幂等写入

这个方案在数据湖和离线数仓里用得最多。核心思想是：以分区为单位，每次写入先覆盖整个分区。

具体做法：

1. 将数据按时间（天/小时）或业务维度分区
2. 写入时，先删除目标分区的所有数据
3. 再写入新的数据
4. 如果写入失败，分区是空的，下次重试重新写入即可

在Hive或Spark SQL里：

-- 先删除分区
ALTER TABLE dwd_orders DROP IF EXISTS PARTITION (dt='2024-01-01');

-- 再写入新数据
INSERT OVERWRITE TABLE dwd_orders PARTITION (dt='2024-01-01')
SELECT * FROM tmp_orders WHERE dt = '2024-01-01';

这个方案的优势是：

• 天然幂等——同一个分区无论写多少次，结果都一样
• 性能好——批量删除+批量写入，比逐行UPSERT快得多
• 适合离线场景——T+1的数据处理，完全够用

缺点也很明显：

• 粒度太粗——只能按分区做幂等，不能精确到行
• 删除操作有风险——如果删除成功但写入失败，数据就丢了
• 不适合实时场景——频繁覆盖分区会影响查询

3.4 三种方案对比与选型

说了这么多，到底选哪个？我整理了一个对比表，方便你决策：

维度	唯一约束方案	外部存储方案	分区覆盖方案
实现复杂度	低	中	低
实时性	高	高	低（适合离线）
存储开销	无额外开销	需要Redis/其他存储	无额外开销
引擎依赖	强（依赖引擎特性）	弱（通用方案）	中（依赖分区能力）
数据粒度	行级	行级	分区级
典型场景	实时写入、CDC同步	多引擎混合、跨系统	离线ETL、全量刷新

我个人建议的选型思路：

• 如果OLAP引擎原生支持UPSERT（如Doris、ClickHouse的ReplacingMergeTree），优先用方案一
• 如果数据源多、引擎杂，或者需要跨系统保证幂等，用方案二
• 如果是离线数仓、T+1场景，方案三最简单也最可靠