Flink SQL Deduplication用 ROW_NUMBER 做流式去重-育师

1. Deduplication 是什么，为什么流式场景尤其需要

Deduplication（去重）是在一组列（去重键）上移除重复行，只保留第一条或最后一条记录。典型原因是：上游 ETL 不是端到端 exactly-once，故障恢复/重试可能把同一业务事件写了多次，导致下游 SUM/COUNT 等统计被“重复行”污染，所以需要在进一步分析前先做去重。(Apache Nightlies)

从实现角度看，Flink SQL 的去重本质上就是 Top-N 的特例：N = 1，并且按处理时间或事件时间排序。(Apache Nightlies)

2. Flink SQL 去重的标准写法（优化器识别模式）

2.1 QUALIFY 写法（更简洁）

官方给出的 Deduplication 语法是：对去重键PARTITION BY，按时间属性ORDER BY，保留ROW_NUMBER() = 1。(Apache Nightlies)

SELECT[column_list]FROMtable_name QUALIFY ROW_NUMBER()OVER(PARTITIONBYcol1[,col2...]ORDERBYtime_attr[ASC|DESC])=1;

2.2 子查询 + WHERE rownum = 1（更通用，适配更多版本）

很多生产环境更习惯用这版（逻辑更直观），并且必须包含rownum = 1，否则优化器可能无法把它翻译成 Deduplicate 算子。(Apache Nightlies)

SELECT[column_list]FROM(SELECT*,ROW_NUMBER()OVER(PARTITIONBYcol1[,col2...]ORDERBYtime_attr[ASC|DESC])ASrownumFROMtable_name)WHERErownum=1;

3. 参数语义：去重键、保留第一条/最后一条、必须是“时间属性”

3.1 PARTITION BY：去重键（deduplicate key）

PARTITION BY col1[, col2...]就是你判定“重复”的那组字段（比如 order_id、message_id、(user_id, item_id) 等）。(Apache Nightlies)

3.2 ORDER BY time_attr：决定保留第一条还是最后一条

排序列必须是时间属性（processing time 或 event time）。(Apache Nightlies)

ORDER BY time_attr ASC：保留最早的一条（first row）
ORDER BY time_attr DESC：保留最新的一条（last row）(Apache Nightlies)

3.3 处理时间 vs 事件时间：结果是否“可复现”

经验上强烈建议优先用事件时间（rowtime）做去重排序：
一些调优/最佳实践文档也明确指出：按 processing time 去重，结果会随运行时机波动；按 event time 去重，结果更确定、更可复现。(Ververica 文檔)

4. 示例：按 order_id 去重，只保留第一次出现（你给的 Orders 例子）

CREATETABLEOrders(order_id STRING,userSTRING,product STRING,numBIGINT,proctimeASPROCTIME())WITH(...);-- 按 order_id 去重，保留第一次出现（ASC）SELECTorder_id,user,product,numFROM(SELECT*,ROW_NUMBER()OVER(PARTITIONBYorder_idORDERBYproctimeASC)ASrow_numFROMOrders)WHERErow_num=1;

这个模式正是 Flink 官方去重示例中强调的写法：ROW_NUMBER + 分区键 + 时间排序 +rownum = 1。(Apache Nightlies)

5. 去重的 Changelog 语义：为什么下游可能会看到 UPDATE/撤回

流式去重不是“简单过滤一次”——当更早/更晚的记录到来时（尤其事件时间乱序），去重结果可能需要被修正，因此会产生changelog（更新流）。

在 Flink SQL 的 retract 机制里，更新通常会被拆成DELETE(-U)+INSERT(+U)两条事件（先撤回旧值，再发新值）。(Confluent)

工程结论：

你的 sink/下游必须能消费更新流（upsert/retract），否则结果会错或写不进去。

6. Sink 选型建议：优先 Upsert 型（尤其写 Kafka / 外部库）

如果你把去重结果写入 Kafka，官方upsert-kafka明确支持消费 changelog：会把 INSERT/UPDATE_AFTER 写成正常消息，把 DELETE 写成 tombstone。(Apache Nightlies)

落外部存储（JDBC、KV、OLAP）时，同样建议使用Upsert 语义（主键一致），让“同一 key 的最新结果”被覆盖更新。

7. 状态与 TTL：去重会“记住 key”，不设 TTL 可能越跑越大

去重需要维护“某个 key 当前保留的是哪一条”，因此会占用 state。Flink Table/SQL 提供table.exec.state.ttl（Idle State Retention Time）用来控制 key 的状态多久没更新就清理。(Apache Nightlies)

注意：TTL 会影响正确性（清理太早可能把还会来的迟到数据当成“新 key”），一般需要结合业务延迟/乱序程度来定。

8. 进阶：Window Deduplication（每个窗口内去重）

如果你需要“每个窗口内按 key 去重”（例如 10 分钟内同一用户只保留最后一次点击），Flink 还有专门的Window Deduplication形态：同样要求ORDER BY是时间属性，且rownum = 1/<=1/<2这种固定谓词，才能让优化器识别。(Apache Nightlies)