Flink SQL Top-N 深度从“实时榜单”到“少写点数据”

1. Top-N 到底是什么？为什么流式 Top-N 更难

Top-N：按某些排序列（比如 sales DESC）取前 N 条（或后 N 条）。既支持 batch，也支持 streaming。(Confluent 文件)

难点在 streaming：

• 数据不断到来、聚合不断变化 →排名随时会变化
• Flink 为了保证结果“永远正确”，会输出更新（UPDATE）/回撤（retraction）给下游，而不是只吐一次结果。(Confluent 文件)

因此：Top-N 的 sink 选型和主键设计，决定了你这条 SQL 能不能跑稳、跑快。

2. Flink SQL Top-N 的标准写法：ROW_NUMBER + OVER + 过滤条件

Flink 用一个固定模式让优化器识别“这是 Top-N”，核心就是：

• ROW_NUMBER() OVER (PARTITION BY ... ORDER BY ...) AS rownum
• 外层必须过滤：rownum <= N
• 其它条件只能用AND拼在一起
• 模式必须严格匹配，否则优化器无法翻译成 TopN 算子(Confluent 文件)

典型模板（传统写法）(Confluent 文件)：

SELECT[column_list]FROM(SELECT[column_list],ROW_NUMBER()OVER([PARTITIONBYcol1[,col2...]]ORDERBYcol1[asc|desc][,col2[asc|desc]...])ASrownumFROMtable_name)WHERErownum<=N[ANDother_conditions];

2.1 你文本里出现的 QUALIFY：Flink 2.0 起更简洁

从 Flink 2.0 开始，SQL 新增了QUALIFY，用来更简洁地过滤窗口函数输出（包括 Top-N / Dedup 这类模式）。(flink.apache.org)

对应写法会更像你贴的那段：

SELECT[column_list],ROW_NUMBER()OVER(PARTITIONBYcategoryORDERBYsalesDESC)ASrownumFROMShopSales QUALIFY rownum<=5;

如果你线上集群版本 < 2.0，就用前面“子查询 + WHERE rownum <= N”的写法；>=2.0 可以优先用 QUALIFY（更短、更不容易写错外层 SELECT）。

3. 连续 Top-N vs 窗口 Top-N：一个“实时滚动榜”，一个“到点出榜”

很多人把 Top-N 都当成一类，其实 Flink 里常见是两种形态：

3.1 连续 Top-N（Continuous Top-N）

• 结果会持续更新：榜单变化就发更新/回撤
• 适合“实时榜单、实时风控榜”这类看“当前最新”的场景
• 你给的ShopSalesTop5 per category 就是典型连续 Top-N (Ververica)

3.2 窗口 Top-N（Window Top-N）

• 窗口结束时才输出最终 Top-N（不发中间更新）
• 状态到期会清理，通常性能更好（因为不需要每条数据都维护“实时榜单”）
• 需要PARTITION BY里包含window_start、window_end，否则优化器无法翻译(Apache Nightlies)

窗口 Top-N 的语法形态（仍然是 Top-N 模式）：(Apache Nightlies)

SELECT*FROM(SELECT*,ROW_NUMBER()OVER(PARTITIONBYwindow_start,window_end,key_colORDERBYmetricDESC)ASrownumFROM(-- Windowing TVF / Window Agg 的结果))WHERErownum<=3;

4. 结果更新语义：为什么你的下游会收到一堆 UPDATE/撤回

连续 Top-N 是Result Updating：Flink 会按排序键维护 TopN 状态；一旦 Top N 发生变化，就会把变化以 retraction/update 形式发下游。(Confluent 文件)

工程影响：

1. 下游必须能“更新”而不是只追加（append-only）
   例如 Upsert-Kafka / JDBC Upsert / 支持主键更新的存储更合适。(Confluent 文件)

2. 结果表必须有正确的唯一键（unique key）
   Top-N 的唯一键通常是：partition columns + rownum，并且还可能继承上游的唯一键。(Confluent 文件)

5. 主键与唯一键：Top-N 正确落库的关键

你给的原文里有个非常重要但经常被忽略的点：

• Top-N 的 unique key =PARTITION BY列 +rownum列
• 同时，Top-N 也可能继承上游 unique key（例如product_id）(Confluent 文件)

这会直接决定你 sink 表怎么建主键。

示例：每个品类实时 Top5（按 sales DESC）

CREATETABLEShopSales(product_id STRING,category STRING,product_name STRING,salesBIGINT)WITH(...);SELECT*FROM(SELECT*,ROW_NUMBER()OVER(PARTITIONBYcategoryORDERBYsalesDESC)ASrow_numFROMShopSales)WHERErow_num<=5;

如果ShopSales的唯一键是product_id，那么 Top-N 这张动态表的唯一键可能同时包含：

• [category, row_num]
• [product_id](Confluent 文件)

落外部存储时的经验法则：

• 如果你要把“榜单结果”落库让别人查：通常用[category, row_num]（表示榜单位置）
• 如果你更关心“每个商品一行、随时更新其排名/指标”：倾向用product_id（表示实体唯一）

6. 性能杀手与经典优化：No Ranking Output Optimization（别把 rownum 写出去）

原始 Top-N 会把rownum作为唯一键的一部分写到结果表里，这可能导致更新风暴：

某条原本排第 9 的记录，突然涨到第 1，那么第 1~9 的所有记录都要作为更新输出一次。(Confluent 文件)

优化方法：外层 SELECT 不输出 rownum（只保留业务列），让消费端自己排序展示。

-- omit row_num field from the outputSELECTproduct_id,category,product_name,salesFROM(SELECT*,ROW_NUMBER()OVER(PARTITIONBYcategoryORDERBYsalesDESC)ASrow_numFROMShopSales)WHERErow_num<=5;

这样通常能显著减少写外部系统的 IO。(Confluent 文件)

7. Streaming 模式下“必须注意”的落库条件

要把 Top-N 输出到外部存储并保证结果正确，外部表需要具备与 Top-N 一致的唯一键/主键语义（至少要支持 upsert）。(Confluent 文件)

如果你采用了“外层不输出 rownum”的优化，一般会更倾向用业务主键（例如product_id）作为外部表主键——这样 Top-N 的变化会以“更新同一行”的方式呈现，更容易被 OLAP/服务层消费。

8. 再给你 3 个生产级调优点（很实用）

8.1 开启/调大 TopN state cache

TopN 有状态缓存，PARTITION BY键数量很大时，缓存命中率会很低 → 性能会掉。可以调大：

table.exec.rank.topn-cache-size: 200000

并结合并行度、N、分区键数量估算命中率。(Ververica 文檔)

8.2 PARTITION BY 里引入“时间字段”，避免 TTL 导致“排名乱序”

在有 TTL 的情况下，如果分区键粒度过粗，状态过期会引发结果异常/乱序。一个常见做法是把“天/小时”等时间字段纳入分区。(Ververica 文檔)

8.3 选对形态：不需要“实时滚动榜”就用 Window Top-N

Window Top-N 在窗口结束时才输出最终结果，通常比连续 Top-N 更省资源。(Apache Nightlies)

9. 常见坑清单（写之前扫一眼，能省半天）

• 忘了写rownum <= N：优化器不识别 Top-N。(Confluent 文件)
• rownum <= N跟其它条件用OR连接：优化器不翻译（必须 AND）。(Confluent 文件)
• Window Top-N 忘了把window_start/window_end放进PARTITION BY：翻译失败。(Apache Nightlies)
• 下游用 append-only sink（只插不更）：遇到 retraction/update 直接崩或结果错。(Confluent 文件)
• 把rownum写进外部表且键设计不当：更新风暴 + 写放大。(Confluent 文件)

10. 小结

• Flink Top-N 本质是：ROW_NUMBER + OVER + rownum <= N的固定模式（为了让优化器识别并生成 TopN 算子）。(Confluent 文件)
• 连续 Top-N 会产生更新/回撤；sink 必须支持 upsert，并且要认真设计唯一键。(Confluent 文件)
• “No Ranking Output Optimization”是生产必备：外层不输出 rownum，大幅降低 IO。(Confluent 文件)
• Flink 2.0 起可以用QUALIFY更优雅地写 Top-N。(flink.apache.org)