当前位置：首页 > news >正文

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

news 2026/4/5 14:08:53

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

1. Window Deduplication 是什么

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

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

直觉理解：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 (
PARTITION BY window_start, window_end [, col_key1...]
ORDER BY time_attr [ASC|DESC]
) AS rownum
FROM table_name  -- 必须是应用了 Windowing TVF 的关系
)
WHERE (rownum = 1 OR rownum <= 1 OR rownum < 2)
[AND other_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 (
SELECT
bidtime, price, item, supplier_id, window_start, window_end,
ROW_NUMBER() OVER (
PARTITION BY window_start, window_end
ORDER BY bidtime DESC
) AS rownum
FROM TABLE(
TUMBLE(TABLE Bid, DESCRIPTOR(bidtime), INTERVAL '10' MINUTES)
)
)
WHERE rownum <= 1;

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

SELECT *
FROM (
SELECT
bidtime, price, item, supplier_id, window_start, window_end,
ROW_NUMBER() OVER (
PARTITION BY window_start, window_end, item
ORDER BY bidtime DESC
) AS rownum
FROM TABLE(
TUMBLE(TABLE Bid, DESCRIPTOR(bidtime), INTERVAL '10' MINUTES)
)
)
WHERE rownum = 1;