Flink SQL Window Deduplication按窗口“保留第一条/最后一条”记录（Streaming）

1. Window Deduplication 是什么

Window Deduplication 是 Deduplication（去重）的窗口版本：在每个窗口（window）+ 分区键（partition keys）内，对重复行做去重，只保留第一条或最后一条。(nightlies.apache.org)

它和“连续表去重（Regular Deduplicate）”最大的区别在于：

• 连续去重：结果可能随每条新记录到来而不断更新；
• 窗口去重：不输出中间结果，只在窗口结束时输出最终结果，并且窗口结束后会清理中间状态，所以如果你不需要“逐条更新”，窗口去重往往更省资源、吞吐更好。(nightlies.apache.org)

直觉理解：Window Deduplication ≈ Window Top-N 的特例（N=1），同样基于ROW_NUMBER()。(nightlies.apache.org)

2. 必须满足的优化器识别条件（写错就不会翻译成 Window Dedup）

2.1 必须来自 Windowing TVF 的结果（包含 window_start/window_end）

窗口去重通常直接跟在 Windowing TVF（TUMBLE/HOP/CUMULATE/SESSION）后使用；Windowing TVF 会生成window_start、window_end、window_time三列。(nightlies.apache.org)

2.2 PARTITION BY 必须包含 window_start 和 window_end

PARTITION BY必须包含window_start, window_end（再加你的业务分区键），否则优化器无法翻译为 Window Deduplication。(nightlies.apache.org)

2.3 WHERE 必须是这三种谓词之一：=1/<=1/<2

窗口去重对过滤谓词也很“死板”：必须是rownum = 1、rownum <= 1或rownum < 2，否则优化器不认。(nightlies.apache.org)

3. 标准语法模板（直接套用）

(nightlies.apache.org)

SELECT[column_list]FROM(SELECT[column_list],ROW_NUMBER()OVER(PARTITIONBYwindow_start,window_end[,col_key1...]ORDERBYtime_attr[ASC|DESC])ASrownumFROMtable_name-- 必须是应用了 Windowing TVF 的关系)WHERE(rownum=1ORrownum<=1ORrownum<2)[ANDother_conditions];

4. 参数语义：保留“第一条”还是“最后一条”

4.1 PARTITION BY：窗口 + 去重键

• window_start, window_end：代表“在每个窗口内”
• col_key1...：你真正的去重键（例如order_id/user_id/item等）(nightlies.apache.org)

4.2 ORDER BY time_attr：决定保留第一条/最后一条

排序字段必须是时间属性。窗口去重里：

• ASC：保留窗口内最早的一条（first）
• DESC：保留窗口内最晚的一条（last）(nightlies.apache.org)

5. 示例：每 10 分钟窗口保留“最后一条记录”

下面这个例子与你给的示例一致：对每个 10 分钟 TUMBLE 窗口，按事件时间bidtime倒序取rownum <= 1，就保留了每个窗口内最后一条。(nightlies.apache.org)

SELECT*FROM(SELECTbidtime,price,item,supplier_id,window_start,window_end,ROW_NUMBER()OVER(PARTITIONBYwindow_start,window_endORDERBYbidtimeDESC)ASrownumFROMTABLE(TUMBLE(TABLEBid,DESCRIPTOR(bidtime),INTERVAL'10'MINUTES)))WHERErownum<=1;

如果你想“每个窗口内按 item 去重，只保留最后一条”，只要把 item 加进分区键：

SELECT*FROM(SELECTbidtime,price,item,supplier_id,window_start,window_end,ROW_NUMBER()OVER(PARTITIONBYwindow_start,window_end,itemORDERBYbidtimeDESC)ASrownumFROMTABLE(TUMBLE(TABLEBid,DESCRIPTOR(bidtime),INTERVAL'10'MINUTES)))WHERErownum=1;

6. 限制与版本注意点（非常重要）

6.1 跟在 Windowing TVF 后时：暂不支持 Session Window

目前如果 Window Deduplication 直接跟在 Windowing TVF 后，TVF 只能是TUMBLE / HOP / CUMULATE，不支持 SESSION；Session window 预计后续支持。(nightlies.apache.org)

6.2 ORDER BY 的时间属性限制：当前要求事件时间

当前 Window Deduplication 的ORDER BY需要是事件时间属性（event time），处理时间（processing time）排序“未来会支持”。(nightlies.apache.org)

7. 生产踩坑清单（写之前对一下）

1. PARTITION BY忘了带window_start, window_end→ 优化器不翻译。(nightlies.apache.org)
2. WHERE写成rownum < 1或rownum = 0等 → 不在认可谓词集合里。(nightlies.apache.org)
3. ORDER BY不是时间属性 / 或用了 processing time（当前限制）→ 计划不通过或无法翻译。(nightlies.apache.org)
4. 事件时间字段没有正确 watermark → 窗口无法按预期关闭（表现为“迟迟不出结果”）；Windowing TVF 对时间属性有明确要求。(nightlies.apache.org)