大数据流处理批处理数据工程【免费下载链接】flink项目地址https://gitcode.com/gh_mirrors/fli/flink点击查看免费下载在实时数据分析场景中每个维度下的 Top N如每个品类的实时销量前五、每个城市的活跃用户前十是最常见的一类查询需求。本文以 Apache Flink 官方文档 Top-N 为主线完整梳理 Top-N 查询的标准 SQL 语法模式与参数规范、流模式下的 Result Updating 行为与唯一键unique key约束、不输出排名列的写放大优化并结合当前仓库中flink-table的优化器规则与运行时算子源码解释 Flink 是如何识别、改写并执行这类查询的。读完本文你可以直接写出可运行的 Top-N SQL理解其回撤语义与 Sink 表设计要点并能在源码层面定位每个执行细节的出处。一、Top-N 查询是什么Top-N 查询要求按若干列排序后取出最小的 N 条或最大的 N 条记录。Flink 文档明确说明最小值集合与最大值集合都算作 Top-N 查询Both smallest and largest values sets are considered Top-N queries。这类查询的典型用途是在批表或流表上按某种条件只展示最底部或最顶部的 N 条记录结果集可继续用于下游分析。Flink 用OVER 窗口子句 过滤条件的组合来表达 Top-N 查询。借助 OVER 窗口的PARTITION BY子句Flink 还支持分组 Top-N——即每个分组各自产出一份 Top-N 结果。文档给出的示例需求是实时取每个品类中销量最高的前五个商品the top five products per category that have the maximum sales in realtime。Top-N 查询同时支持批表和流表上的 SQL。二、Top-N 的标准语法与参数规范Top-N 语句必须遵循如下语法模式SELECT [column_list] FROM ( SELECT [column_list], ROW_NUMBER() OVER ([PARTITION BY col1[, col2...]] ORDER BY col1 [asc|desc][, col2 [asc|desc]...]) AS rownum FROM table_name) WHERE rownum N [AND conditions]参数说明完整继承原文档参数说明ROW_NUMBER()按照分区内行的顺序给每行分配一个从 1 开始的唯一序号。当前仅支持ROW_NUMBER作为 over 窗口函数未来计划支持RANK()和DENSE_RANK()。PARTITION BY col1[, col2...]指定分区列。每个分区都会有一份独立的 Top-N 结果省略该子句时整张表作为单一分区对应全局 Top-N。ORDER BY col1 [asc\|desc][, col2 [asc\|desc]...]指定排序列不同列可以有不同的升降序方向。WHERE rownum N必需条件。rownum N是 Flink 识别该查询为 Top-N 查询的标志N 表示保留最小或最大的前 N 条记录。[AND conditions]可以任意追加其他过滤条件但其他条件只能与rownum N使用AND连接。文档中有一条强约束提示上述模式必须被精确遵循否则优化器无法将该查询翻译为 Top-N 执行计划the above pattern must be followed exactly, otherwise the optimizer wont be able to translate the query。也就是说把WHERE rownum N写成WHERE rownum N1、或在子查询外层套多余的列改写都会导致退化为普通窗口函数 过滤的执行路径。从源码结构看这一模式识别发生在逻辑计划转换阶段Calcite 的LogicalRank节点经由 FlinkLogicalRankConverter 转换为 Flink 自己的 Rank 逻辑节点节点上携带partitionKey分区列、orderKey排序方向、rankType排名函数类型、rankRange排名范围即rownum N解析出的 [1, N] 区间以及outputRankNumber是否需要在输出中保留排名列等属性见 FlinkLogicalRank。这正是文档中必须精确遵循模式的底层原因——只有模式匹配成功这些属性才能被正确提取并驱动专用算子的选择。另外文档中当前只支持ROW_NUMBER未来支持RANK()与DENSE_RANK()的说法在运行时源码中同样得到印证RankType 枚举定义了ROW_NUMBER、RANK、DENSE_RANK三种类型而流式 TopN 算子基类 AbstractTopNFunction 的构造逻辑中RANK与DENSE_RANK分支会抛出UnsupportedOperationException错误信息为 RANK() on streaming table is not supported currently目前只有ROW_NUMBER能顺利通过。三、流模式下的 Result Updating 语义文档对 Top-N 查询有一个关键行为提示TopN 查询是 Result Updating 查询。Flink SQL 会按排序键对输入数据流排序如果 Top N 记录发生了变化变化的记录会以下游可见的回撤/更新记录retraction/update records发出。建议使用支持更新的存储作为 Top-N 查询的 Sink。此外如果 Top N 结果需要落到外部存储结果表必须与 Top-N 查询具有相同的唯一键。这意味着流式 Top-N 的输出不是一次性的追加结果而是会随数据到来不断修正当一个新记录挤进 Top N、把某条原第 N 名的记录挤出时被挤出的记录会以删除/回撤消息发出新进入的记录以插入消息发出。因此Append-only 的 Sink如普通文件写出会得到错误结果Sink 必须能按唯一键处理 UPDATE/DELETE 消息例如 JDBC 表、Kafka upsert 等支持更新的存储Sink 表的唯一键必须与 Top-N 查询推导出的唯一键一致否则更新与回撤无法正确落到对应行。Top-N 查询的唯一键推导规则文档给出的规则是Top-N 查询的唯一键是分区列与 rownum 列的组合同时Top-N 查询还可以继承上游表的唯一键。文档以如下作业为例假设product_id是ShopSales表的唯一键那么该 Top-N 查询的唯一键就同时包括[category, rownum]分区列 rownum和[product_id]继承自上游。这一推导上游唯一键的能力在源码中有明确落点StreamExecRank 在构造算子时会传入rowKeySelector行唯一键选择器等参数用于把上游唯一键一并纳入算子的输出从而让下游 Sink 按正确的键做更新。四、完整示例实时取每个品类销量前五以下是文档给出的流表 Top-N 示例即实时取每个品类中销量最高的前五个商品CREATE TABLE ShopSales ( product_id STRING, category STRING, product_name STRING, sales BIGINT ) WITH (...); SELECT * FROM ( SELECT *, ROW_NUMBER() OVER (PARTITION BY category ORDER BY sales DESC) AS row_num FROM ShopSales) WHERE row_num 5逐行解读ShopSales是流式源表WITH (...)为具体连接器参数可按所用连接器填写内层查询通过ROW_NUMBER() OVER (PARTITION BY category ORDER BY sales DESC)按category分组、按sales降序编号row_num为 1 表示该品类当前销量第一外层WHERE row_num 5过滤出每个品类的前 5 名按第二节的唯一键规则该查询的唯一键为[category, row_num]若ShopSales声明了product_id唯一键则[product_id]也是唯一键之一。五、No Ranking Output Optimization不输出排名列以减少写放大文档专门用一节No Ranking Output Optimization讲解了流模式下 Top-N 的一个重要优化问题如上所述rownum字段会作为唯一键的一部分写入结果表这可能带来大量写入。举例当排名第 9 的记录如product-1001被更新、排名跃升到第 1 时排名第 19 的所有记录都会以更新消息的形式输出到结果表。如果结果表接收的数据量过大它本身就会成为整个 SQL 作业的瓶颈。优化方式在 Top-N 查询的外层 SELECT 中省略 rownum 字段。这是合理的因为 Top N 的记录数量通常不大消费端可以自行快速排序。省略 rownum 后上面例子中只需要把真正变化的记录product-1001发给下游从而大幅减少对结果表的 IO。文档给出的优化后写法CREATE TABLE ShopSales ( product_id STRING, category STRING, product_name STRING, sales BIGINT ) WITH (...); -- omit row_num field from the output SELECT product_id, category, product_name, sales FROM ( SELECT *, ROW_NUMBER() OVER (PARTITION BY category ORDER BY sales DESC) AS row_num FROM ShopSales) WHERE row_num 5文档对此还有一个醒目的注意事项Attention in Streaming Mode为了把上述查询写入外部存储并得到正确结果外部存储必须与 Top-N 查询具有相同的唯一键。以上面这个查询为例如果product_id是查询的唯一键那么外部表也必须以product_id作为唯一键。在源码层面是否输出排名列正是第二节日志节点中outputRankNumber属性的职责运行时算子基类 AbstractTopNFunction 的createOutputRow方法会依据outputRankNumber决定输出行是在原行后面拼接 rank 值还是直接输出不含 rank 的行。以 AppendOnlyTopNFunction 为例处理逻辑中明确区分了processElementWithRowNumber需要输出排名列或带 offset 时使用与processElementWithoutRowNumber无排名输出的轻量算法两条路径——后者正是不输出排名列优化在算子内部的实现落点。六、源码级机制Top-N 的运行时算子家族文档描述的是 SQL 层面的语法与语义而当前仓库中flink-table模块的源码展示了这一语义在运行时是如何落地的可以作为深入理解的佐证。6.1 算子选择同一 SQL 模式多套运行时实现流式执行节点 StreamExecRank 会依据输入流的 changelog 模式追加 / 仅更新 / 可回撤和排名范围选择具体的运行时函数。从源码结构看它构造的候选包括AppendOnlyFirstNFunction输入为 insert-only 且带 offset如rownum 1 AND rownum N场景FastTop1Functionrownum 1的 Top-1 特化实现只需比较当前记录是否比榜首更好无需维护完整 N 条有序缓冲AppendOnlyTopNFunction输入为 insert-only 流的标准 Top-N 实现UpdatableTopNFunction输入含 update 消息如按主键更新时的实现RetractableTopNFunction输入含 retract/delete 消息时的实现。这些算子都继承自基类 AbstractTopNFunction它是一个 FlinkKeyedProcessFunction——这与文档每个分区各有一份 Top-N 结果的语义一一对应分区列即 KeyBy 的 key每个分区的排名状态隔离维护天然支持并行执行。6.2 状态、缓存与指标以 AppendOnlyTopNFunction 为例可以看清 Top-N 算子的资源结构dataStateMapState以排序键为 key保存 Top-N 缓冲区内各排序键对应的记录列表是参与 checkpoint 的持久状态bufferTopNBuffer内存堆结构是dataState中 Top-N 数据的镜像用于加速比较与插入kvSortedMapLRU Cache以分区键为 key 缓存各分区的 buffer避免每次访问都从状态反序列化恢复打开算子时日志会打印 Top{N} operator is using LRU caches key-size。基类还暴露了可观测指标AbstractTopNFunction 的registerMetrictopn.cache.hitRateLRU 缓存命中率与topn.cache.size缓存占用行数运维时可用它们判断分区热点与缓存配置是否合理。此外当排名上界 N 为变量由数据列决定且前后取值不一致时topn.invalidTopSize计数器会累加见 AbstractTopNFunction#L158说明该 Top-N 的 N 值不稳定、可能影响结果正确性。值得说明的是这些运行时细节缓存、FastTop1、按 changelog 模式分派算子属于源码实现事实官方语法文档并未承诺它们但它们与文档描述的模式精确匹配Result Updating唯一键对齐等行为规范是同一套机制的两面。七、批模式与流模式的支持差异文档指出 Top-N 查询同时支持批表和流表上的 SQLTop-N queries are supported for SQL on batch and streaming tables两者的差异主要体现在语义层面批模式数据完整一次性排序截取前 N 条即可不存在结果更新问题输出是确定性的结果集流模式如第三节所述输出是 Result Updating 流必须关注回撤消息、Sink 的可更新性与唯一键一致性并且当前流表仅支持ROW_NUMBER排名函数RANK()/DENSE_RANK()尚未支持这一点在 AbstractTopNFunction 中通过异常直接拦截。八、实践要点小结精确套用语法模式ROW_NUMBER() OVER (PARTITION BY ... ORDER BY ...) AS rownum 外层WHERE rownum Nrownum N是触发 Top-N 优化的必要条件其他过滤条件只能用AND与其组合。流模式 Sink 必须支持更新且 Sink 表唯一键必须与 Top-N 推导出的唯一键分区列 rownum或继承的上游唯一键一致。不需要排名列时就省略它外层 SELECT 不选 rownum可以显著减少对结果表的写放大这是官方文档明确推荐的优化。流表当前只有ROW_NUMBER可用不要期望RANK()/DENSE_RANK()源码中直接抛不支持异常。遇到性能问题时可结合算子自带的topn.cache.hitRate、topn.cache.size、topn.invalidTopSize等指标定位缓存命中与 N 值稳定性问题。围绕本文主题的延伸阅读与验证路径语法文档 docs/content/docs/dev/table/sql/queries/topn.md、逻辑节点 FlinkLogicalRank、执行节点 StreamExecRank、运行时算子目录 flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/operators/rank 及其测试 UpdatableTopNFunctionTest。赞分享大数据流处理批处理数据工程【免费下载链接】flink项目地址https://gitcode.com/gh_mirrors/fli/flink点击查看免费下载相关推荐Flink Window Top-N 完整指南窗口内 Top-N 查询的语法、原理与实战Flink Window Top N 完整指南窗口内 Top N 查询的语法、原理与实战 导读 Window Top N窗口 Top N是 Flink S大数据流处理批处理数据工程Flink SQL 窗口 Top-N 查询完全指南语法、示例与底层实现原理Flink SQL 窗口 Top N 查询完全指南语法、示例与底层实现原理 窗口 Top N 是 Flink SQL 中一类特殊的 Top N 查询它针对每大数据流处理批处理数据工程Flink SQL Top-N 查询完全指南语法详解、结果更新语义与性能优化Flink SQL Top N 查询完全指南语法详解、结果更新语义与性能优化 Top N 查询是 Flink SQL 中按指定列排序后取前 N 个最小值或最大大数据流处理批处理数据工程上一篇如何为Translumo配置代理避免IP封禁的终极指南下一篇B站视频下载终极指南轻松保存4K大会员和充电专属内容创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考