订阅更新

主键唯一性，光靠PRIMARY KEY是不够的——ETL层去重的5种姿势

数据质量手册 第 3 题 约 11 分钟读完

本文目录
  1. 先说结论
  2. 为什么数仓没有真正的PRIMARY KEY？
  3. 姿势一：ROW\NUMBER() 取最新版本（最常用）
  4. 姿势二：INSERT OVERWRITE 幂等全量重写
  5. 姿势三：MERGE INTO 增量合并（数据湖时代的标配）
  6. 姿势四：Flink的Dedup算子（实时场景）
  7. 姿势五：业务幂等键 + UPSERT（最优雅的方案）
  8. 五种姿势对比
  9. 面试高频追问
  10. 实战踩坑：补数重跑导致全表数据翻倍
  11. 总结

先说结论

数仓没有真正的主键约束，所以重复必然发生。去重的5种姿势：ROW_NUMBER取最新、INSERT OVERWRITE全量重写、MERGE INTO增量合并、Dedup算子（Flink）、幂等键+UPSERT。选哪种取决于场景：批/流、全量/增量、是否有时间字段判断版本。

最容易翻车的不是去重逻辑本身，而是漏掉了「重复发生在哪一层」——CDC窗口期、Kafka At-Least-Once重发、补数重跑、维度表多版本，每一层都要有对应方案。

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

详细讲解

「主键唯一性」是数据质量里看似最简单的话题，但生产环境我见过太多翻车案例。Hive、MaxCompute、Spark这些数仓引擎根本没有真正的PRIMARY KEY约束（只有逻辑主键），数据重复完全靠ETL层自己保证。一旦上游CDC、消息重发、补数重跑，就会产生重复。今天就把ETL层去重的5种姿势讲清楚，每一种都附场景和代码。

为什么数仓没有真正的PRIMARY KEY？

很多刚转大数据的同学不理解：MySQL有PRIMARY KEY，为什么Hive/MaxCompute没有？答案是：

分布式存储不能强约束唯一性。PRIMARY KEY的本质是「插入时检查是否已存在」，但在分布式存储里，数据被切成多个文件分散存储，插入时全表扫描去重检查代价巨大，性能上无法接受。

-- MaxCompute里看似有PRIMARY KEY，但其实是「Hash Cluster Key」
CREATE TABLE dwd_order (
  order_id BIGINT,
  user_id BIGINT,
  amount DECIMAL(18,2),
  PRIMARY KEY (order_id)  -- 仅用于优化，不强制唯一
) CLUSTERED BY (order_id) SORTED BY (order_id) INTO 32 BUCKETS;
 
-- 你完全可以插入两条相同order_id的记录，引擎不会报错
INSERT INTO dwd_order VALUES (1001, 1, 100), (1001, 1, 200);
SELECT COUNT(*) FROM dwd_order WHERE order_id = 1001;  -- 结果：2

Hologres和Doris等OLAP引擎有真正的Primary Key约束（基于Aggregate模型或Replace策略），但代价是写入性能和存储成本。所以数仓里去重的责任必须由ETL层承担。

姿势一：ROW_NUMBER() 取最新版本（最常用）

最经典、最通用的去重方法。适合「同一主键有多条记录但只取最新一条」的场景，比如CDC同步表、订单状态表。

-- 场景：CDC同步过来的订单表，同一order_id可能有多条变更记录
-- 目标：每个order_id只保留update_time最新的那条
INSERT OVERWRITE TABLE dwd_order PARTITION(dt='${bizdate}')
SELECT order_id, user_id, amount, status, update_time
FROM (
  SELECT *,
    ROW_NUMBER() OVER (
      PARTITION BY order_id
      ORDER BY update_time DESC, op_seq DESC  -- op_seq处理同一秒多条
    ) AS rn
  FROM ods_order_cdc
  WHERE dt = '${bizdate}'
) t WHERE rn = 1;

适用场景：CDC同步、变更流、慢变维快照。

不适用：没有时间字段或版本号无法判断哪条是「最新」时不能用；超大表（10亿+）做ROW_NUMBER会触发全局Shuffle，性能差。

姿势二：INSERT OVERWRITE 幂等全量重写

适合维度表、配置表、小数据量的全量计算。最大的优点是天然幂等——重跑多少次结果都一样。

-- 维度表全量重建：天然不会重复
INSERT OVERWRITE TABLE dim_product PARTITION(dt='${bizdate}')
SELECT product_id, product_name, category, brand, price
FROM ods_product_full
WHERE dt = '${bizdate}';
 
-- 重跑1次和重跑10次结果完全一样，永远是最新的全量

适用场景：维度表、配置表、参数表、小数据量。

不适用：数据量太大、每天只有少量变更的场景——全量重写浪费资源。这种应该用MERGE INTO。

姿势三：MERGE INTO 增量合并（数据湖时代的标配）

Delta Lake、Iceberg、Hudi、Hologres都支持MERGE INTO。是大表增量更新的最优解，比UNION ALL+OVERWRITE性能好70%以上。

-- Delta Lake的MERGE INTO示例（出海硬件公司大量使用）
MERGE INTO dwd_user_profile t
USING (
  SELECT user_id, age, gender, city, last_active_time
  FROM ods_user_change WHERE dt = '${bizdate}'
) s ON t.user_id = s.user_id
WHEN MATCHED AND s.last_active_time > t.last_active_time THEN
  UPDATE SET t.age=s.age, t.gender=s.gender, t.city=s.city,
             t.last_active_time=s.last_active_time
WHEN NOT MATCHED THEN
  INSERT (user_id, age, gender, city, last_active_time)
  VALUES (s.user_id, s.age, s.gender, s.city, s.last_active_time);

我在出海硬件公司把所有维度表从UNION+OVERWRITE改成MERGE INTO，代码量减少70%，单表更新时间从30分钟降到8分钟，主要是因为MERGE INTO只重写变更涉及的文件。

姿势四：Flink的Dedup算子（实时场景）

实时数仓里去重不能用ROW_NUMBER一刀切——Kafka消息至少一次（At-Least-Once）必然有重复，状态又不能无限保留。Flink SQL的Dedup有专门的写法：

-- Flink SQL去重：基于事件时间，仅保留每个key的第一条（去重）
SELECT order_id, user_id, amount, event_time
FROM (
  SELECT *,
    ROW_NUMBER() OVER (
      PARTITION BY order_id
      ORDER BY event_time ASC  -- ASC=保留首条，DESC=保留最新
    ) AS rn
  FROM kafka_order_stream
) WHERE rn = 1;
 
-- 关键：必须配TTL，否则状态会无限膨胀
SET table.exec.state.ttl = '24h';

注意点：Flink Dedup会维护State（key→已见过的flag），State大小=去重Key基数×TTL。如果key基数很大且TTL长，会导致状态爆炸。我在某元宇宙社交应用项目就吃过这个亏——一开始没设TTL，跑了一周后CheckPoint时间从30秒涨到5分钟。

姿势五：业务幂等键 + UPSERT（最优雅的方案）

在源头就保证幂等，比下游去重优雅得多。给每条数据生成业务上的「幂等键」（biz_idempotent_key），下游用UPSERT写入：

-- 上游生成幂等键（订单号 + 操作类型 + 时间）
biz_idempotent_key = CONCAT(order_id, '_', op_type, '_', op_time)
 
-- 下游写Hologres（支持INSERT ON CONFLICT）
INSERT INTO dwd_order_event (biz_idempotent_key, order_id, op_type, ...)
VALUES (...)
ON CONFLICT (biz_idempotent_key) DO UPDATE
  SET op_time = EXCLUDED.op_time, ...;
 
-- 下游写Kafka时也可以用幂等Producer
props.put("enable.idempotence", "true");

五种姿势对比

姿势适用场景性能复杂度幂等性
ROW_NUMBER批量CDC同步中（全局Shuffle）低弱（依赖排序字段）
INSERT OVERWRITE维度表/小表全量高（小数据）/低（大数据）极低强
MERGE INTO大表增量更新高（只改变更文件）中强
Flink Dedup实时流去重中（状态开销）中依赖TTL
UPSERT+幂等键实时/批量都行高高（需源头改造）极强

面试高频追问

追问一：ROW_NUMBER去重数据量太大跑不动怎么办？

第一，先按主键预聚合（GROUP BY），把多版本合并成一行（用MAX_BY等函数取最新），减少Shuffle数据量；第二，开启AQE自动倾斜处理；第三，对超大表考虑改成MERGE INTO增量合并。我在出海硬件公司处理过50亿行的订单变更表，从ROW_NUMBER改成MERGE INTO后从40分钟降到6分钟。

追问二：实时数仓里如何保证Exactly-Once而不仅是去重？

Exactly-Once = At-Least-Once + 去重。Flink层面：开启Checkpoint+幂等Sink；Sink层面：用支持事务的存储（Kafka事务、Hologres、Iceberg）；业务层面：每条数据带幂等键。三层结合才是真正的Exactly-Once，缺一不可。

追问三：发现历史数据已经有重复了怎么补救？

先评估影响：用COUNT(*)-COUNT(DISTINCT pk)看重复量级；再评估方案：少量重复用UPDATE/DELETE修复，大量重复直接重跑分区或全表去重重写。最后一定要做事后预防：补DQC规则、补告警阈值，避免下次再出。

实战踩坑：补数重跑导致全表数据翻倍

有一次出海硬件公司某表数据有问题，我用INSERT INTO（不是OVERWRITE）写补数据，结果全表数据量翻倍——下游报表GMV直接虚高100%。当晚被业务方@到群里。

根因：我执行的SQL是`INSERT INTO target SELECT * FROM source`，本意是补一个分区，但目标表已经有当天分区了，结果就是叠加而不是覆盖。

解决方案：立即用INSERT OVERWRITE重新写入正确分区。事后做了三件事：1）所有补数SQL模板强制用OVERWRITE，禁用INSERT INTO到分区表；2）任务上线前必须review SQL；3）DQC加一条「数据量同比波动告警」P0级别，防止类似问题再次发生。

教训：生产环境的INSERT INTO是危险操作。能用OVERWRITE就用OVERWRITE，能用MERGE就用MERGE。INSERT INTO更适合追加流水类型表，且必须配合幂等键。

总结

数仓没有真正的主键约束，重复必然发生。5种去重姿势按场景选：CDC用ROW_NUMBER、维度小表用OVERWRITE、大表增量用MERGE INTO、实时流用Flink Dedup、最优雅是源头幂等键+UPSERT。面试时把5种姿势+对比表+一个真实事故讲清楚，能秒杀大部分候选人。

下一篇聊一个工程化思维很强的话题：数据校验放事前还是事后？数据质量门禁（CI/CD for Data）怎么落地？

你们生产环境最常用哪种去重方式？踩过最坑的重复事故是什么？评论区聊聊～

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

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

相关题目

  1. 23-06 某指标突增200%是真的还是脏数据？——统计学异常检测落地 数据开发与治理
  2. 23-07 实时数仓和离线数仓数据对不上——流批对账（Reconciliation）框架怎么搭？ 数据开发与治理
  3. 23-08 业务库↔数仓↔BI三方数据对不上——跨系统对账框架怎么落地？ 数据开发与治理
  4. 23-09 数据延迟3天才到——Late Arriving Data怎么处理才不破坏历史？ 数据开发与治理
  5. 23-10 上游分布悄悄变了——数据漂移（Data Drift）你怎么发现？ 数据开发与治理
  6. 23-11 上游schema改了下游全挂——Schema Evolution怎么治？ 数据开发与治理