★ FEATURED ARTICLE 1. 从一次凌晨告警说起ROW_NUMBER 为什么会在 Flink SQL 里翻车凌晨两点被电话叫醒作业挂了日志里一行红字格外刺眼org.apache.flink.table.api.TableException: Cannot generate a valid execution plan for the given query。翻到下面罪魁祸首是 SQL 里那句看起来人畜无害的ROW_NUMBER() OVER (PARTITION BY ... ORDER BY ...)。这个场景我遇到过不止一次而且每次踩坑的人背景都差不多——要么是从 Hive SQL、MySQL 8 转过来的要么是刚上手 Flink SQL 不久觉得窗口函数是 SQL 标配写起来理所当然。先把结论摆在前面ROW_NUMBER()在 Flink SQL 里不是不能用而是它的使用方式和传统批处理数据库有本质区别。Flink SQL 是流批一体的同一段 SQL 在流模式下执行时引擎必须考虑数据无界、状态无限增长、乱序到达这些批处理里根本不存在的问题。ROW_NUMBER()属于Top-N 去重/排名语义Flink 对它的支持是有明确边界条件的一旦越过边界要么直接报错要么作业能起来但状态爆炸、结果不对。这篇内容我打算把这件事彻底讲透报错到底分哪几类、每一类的根因是什么、怎么改、改完怎么验证。适合正在用 Flink SQL 做实时数仓、实时去重、实时排名的同学也适合刚接触 Flink SQL 想搞清楚为什么我的窗口函数跑不起来的初学者。我会尽量用大白话把执行计划、状态、时间属性这些概念串起来让你看完能自己判断一段带ROW_NUMBER()的 SQL 到底能不能跑。需要提前说明的是Flink 版本差异对这个问题的表现影响很大。1.13 之前、1.13 到 1.15、1.16 之后优化器对ROW_NUMBER()的处理策略一直在演进。下面讲的内容以 1.13 及以后的版本为主线涉及版本差异的地方我会单独标出来。如果你还在用更老的版本很多报错可能连提示信息都不一样建议先升级再对照排查。2. 报错不是一种先分清你撞上的是哪堵墙很多人一看到报错就慌了直接去搜错误信息结果搜到的方案五花八门试了一圈还是不行。问题在于ROW_NUMBER()相关的报错其实分好几个层次每一层的根因和解法完全不同。先做分类再对症下药效率会高很多。2.1 第一类语法解析期就挂掉这类报错最直接通常长这样org.apache.calcite.sql.parser.SqlParseException: Encountered OVER at line ...或者SQL validation failed. From line ..., column ...: Cannot apply ROW_NUMBER to arguments of type ...出现这种报错八成是写法本身不符合 Flink SQL 的语法规范。常见原因有几个一是OVER子句里漏了ORDER BYFlink 要求ROW_NUMBER()必须带ORDER BY因为排名必须有明确的排序依据二是PARTITION BY里用了不支持的表达式比如某些函数嵌套三是把ROW_NUMBER()写在了WHERE子句里而窗口函数在 Flink SQL 中只能出现在SELECT列表里不能直接用于过滤。这里有个特别容易踩的点从 MySQL 8 转过来的人习惯写WHERE rn 1但在 Flink SQL 里rn是SELECT里算出来的别名WHERE执行顺序在SELECT之前根本看不到这个别名。正确做法是用子查询包一层在外层过滤。这个逻辑和 Hive 一样但很多人写 Flink SQL 时会忘。2.2 第二类校验通过但生成执行计划失败这类报错就是开头提到的那个Cannot generate a valid execution plan for the given query后面通常会跟一段更具体的原因比如The window function requires the input to be sorted by the partition key and order key.或者Sorting is required but not supported in streaming mode without time attribute.这类报错的核心矛盾在于流模式下Flink 无法对无界数据做全局排序。ROW_NUMBER()的语义要求先按PARTITION BY分组、再按ORDER BY排序然后给每行编号。批处理里数据是有限的排完就完事流处理里数据源源不断你让它怎么排所以 Flink 必须借助时间属性和状态来近似实现这个语义而如果你的 SQL 里没有提供时间属性优化器就不知道怎么把无界流切成有界的排序单元于是直接拒绝生成计划。2.3 第三类能跑但结果不对或状态爆炸这是最阴险的一类因为它不报错。作业正常启动Checkpoint 也能做但跑着跑着你会发现结果里同一个 key 出现了多条本应被去重的记录或者 TaskManager 内存一路飙升最后 OOM。根因通常是没有正确设置状态过期时间state TTL或者**ORDER BY的字段选择不当导致状态无限累积**。ROW_NUMBER()在流模式下Flink 会为每个 partition key 维护一个状态用来记住当前已经见过的记录以便判断新来的记录该排第几。如果 partition key 的基数很大比如按用户 ID 分区有几千万用户或者 order key 一直在变比如按事件时间排序但事件时间持续前进状态就会无限增长。下面这张表可以帮你快速定位自己属于哪一类报错/现象出现阶段核心根因典型解法SqlParseException解析期语法不符合规范补 ORDER BY、改用子查询过滤Cannot generate execution plan计划生成期流模式无法全局排序引入时间属性、改用窗口 TVF结果重复/状态爆炸运行期状态无界增长设置 TTL、调整分区键作业启动即失败提交期版本不支持该写法升级版本或改写 SQL把这三类分清楚后面的排查就不会像无头苍蝇一样乱撞了。3. 流模式下 ROW_NUMBER 的底层逻辑为什么它和批处理不一样要真正理解报错得先搞清楚 Flink 在流模式下是怎么实现ROW_NUMBER()的。这部分稍微偏原理但理解了它你后面改 SQL 就是有的放矢而不是照抄别人的方案。3.1 批处理里的 ROW_NUMBER 是一次性排序在 MySQL、Hive、Spark SQL 的批处理场景里ROW_NUMBER() OVER (PARTITION BY a ORDER BY b)的执行逻辑很直观先把所有数据按a分组组内按b排序然后从 1 开始编号。因为数据是有限的排完序结果就确定了内存不够还能落盘。整个过程是有始有终的。3.2 流处理里它变成了持续维护的排名状态到了 Flink 流模式数据是一条一条来的没有全部数据这个概念。Flink 的做法是为每个PARTITION BY的 key 维护一份状态状态里保存当前这个 key 下已经见过的、参与排名的记录。每来一条新记录就把它和状态里的记录比较重新计算排名然后输出排名变化的结果。这就带来两个直接后果。第一状态大小取决于每个 key 下需要保留多少条记录。如果你要的是rn 1取第一条理论上每个 key 只需要保留当前排名第一的那条状态是可控的但如果你要的是rn 100那每个 key 至少要保留 100 条状态就大了 100 倍。第二排序需要时间维度的界定。如果ORDER BY用的是处理时间proctime那当前是明确的如果用的是事件时间event timeFlink 需要靠 watermark 来判断什么时候之前的记录可以认为到齐了否则它不敢把某条记录判定为最终排名。3.3 时间属性是解锁执行计划的那把钥匙这就是为什么第二类报错总在提time attribute。Flink 的优化器在生成ROW_NUMBER()的执行计划时会检查你的ORDER BY字段是不是时间属性字段。如果是它就能把无界流按时间切成一个个有界的窗口在每个窗口内做排名状态也能随窗口结束而清理。如果不是它就不知道边界在哪只能拒绝。所以你会看到一个规律同样一句ROW_NUMBER()ORDER BY后面跟的是事件时间字段且该字段已用WATERMARK FOR ... AS ...声明就能跑跟的是普通业务字段比如金额、数量大概率报错。这不是 Flink 故意刁难而是流处理的本质决定的。提示这里说的时间属性指的是在CREATE TABLE时用WATERMARK FOR ts AS ts - INTERVAL 5 SECOND这种方式声明过的字段或者用PROCTIME()生成的处理时间字段。普通的TIMESTAMP类型字段如果没有声明 watermark不算时间属性。理解了这一层你再看那些报错信息就不会觉得莫名其妙了。接下来讲具体怎么改。4. 四套可落地的改写方案对应不同业务场景知道了根因改写就有方向了。我把常见的业务场景和对应的改写方案整理成四套你可以直接对号入座。每套方案我都会说清楚适用条件、具体写法、以及为什么这么写。4.1 方案一实时去重取最新一条最常见业务场景上游是 Kafka 里的变更日志同一个订单 ID 可能有多条更新记录你要取每个订单最新状态的那一条往下游发。原始写法会报错SELECT order_id, status, update_time FROM ( SELECT order_id, status, update_time, ROW_NUMBER() OVER (PARTITION BY order_id ORDER BY update_time DESC) AS rn FROM order_changelog ) WHERE rn 1;如果update_time没有声明 watermark这段 SQL 在流模式下大概率报无法生成执行计划。改写方式有两种。第一种如果update_time可以声明为事件时间CREATE TABLE order_changelog ( order_id STRING, status STRING, update_time TIMESTAMP(3), WATERMARK FOR update_time AS update_time - INTERVAL 5 SECOND ) WITH (...);声明之后同样的ROW_NUMBER()查询就能生成计划了。因为 Flink 知道按update_time排序时watermark 之前的记录可以认为到齐状态可以随 watermark 推进而清理。第二种如果业务上其实不关心事件时间只想要当前最新那更推荐用LAST_VALUE或者干脆用GROUP BYMAX的思路避免引入排名状态。比如SELECT order_id, LAST_VALUE(status) AS status, MAX(update_time) AS update_time FROM order_changelog GROUP BY order_id;这种写法状态更小语义也更贴合取最新的需求。很多人一上来就用ROW_NUMBER()其实是因为习惯了批处理思维没想到流模式下有更轻量的等价写法。4.2 方案二按时间窗口取 Top-N业务场景每 5 分钟统计一次各商品销量前 10 名。这种场景的正确姿势是用窗口表值函数Window TVF而不是裸写ROW_NUMBER()。写法如下SELECT window_start, window_end, product_id, sales FROM ( SELECT window_start, window_end, product_id, SUM(amount) AS sales, ROW_NUMBER() OVER (PARTITION BY window_start, window_end ORDER BY SUM(amount) DESC) AS rn FROM TABLE(TUMBLE(TABLE sales_stream, DESCRIPTOR(event_time), INTERVAL 5 MINUTES)) GROUP BY window_start, window_end, product_id ) WHERE rn 10;这里的关键是PARTITION BY里带了window_start, window_end。因为窗口是有界的每个窗口内的数据量有限Flink 就能在每个窗口内安全地做排名窗口结束后状态自动清理。这是流模式下做 Top-N 最标准、最推荐的方式。注意窗口 TVF 是 Flink 1.13 引入的1.13 之前用的是TUMBLE_START那套老语法两者不能混用。如果你在 1.13 之后还用老语法写窗口再套ROW_NUMBER()很容易出现计划生成失败。4.3 方案三用 Deduplication 语法替代手写 ROW_NUMBERFlink SQL 其实内置了去重语法专门用来处理按主键取第一条的场景底层会自动优化成ROW_NUMBER()但状态管理更智能SELECT order_id, status, update_time FROM ( SELECT *, ROW_NUMBER() OVER (PARTITION BY order_id ORDER BY update_time DESC) AS rn FROM order_changelog ) WHERE rn 1;这段看起来和方案一一样但如果你用的是 Flink 1.13并且update_time是时间属性优化器会识别出这是标准的去重模式自动应用Deduplicate算子状态只保留每个 key 的第一条而不是保留所有记录。这就是为什么声明时间属性这么重要——它不只是让计划能生成还能让状态大幅减小。判断是否被识别为去重模式可以看执行计划里有没有Deduplicate节点。用EXPLAIN命令就能看到EXPLAIN SELECT ... ;如果计划里出现Deduplicate(keep[FirstRow], key[order_id], order[update_time])说明优化生效了。4.4 方案四确实需要全局排序时的兜底做法有些业务场景就是需要跨时间的全局排名比如累计销售额排名。这种在流模式下本质上是有状态、无界的Flink 不会给你一个完美的方案只能做近似。常见做法是用处理时间PROCTIME()作为排序依据接受排名随时间变化的语义设置合理的状态 TTL比如table.exec.state.ttl 24h让状态不会无限增长如果业务能接受把全局排名改成按天窗口排名用方案二解决。我个人的经验是绝大多数号称需要全局排名的需求拆解之后都能落到窗口排名上。真正需要无界全局排名的场景极少而且那种场景下批处理或离线计算往往更合适。硬要在流里做全局排名最后大概率是状态爆炸。5. 排查链路实录一次状态爆炸的完整定位过程光讲方案不够我拿一次真实的排查过程来演示让你看到遇到问题时应该按什么顺序去查。这个案例是某实时看板作业用了ROW_NUMBER()做去重跑了两天之后 TaskManager 内存告警。5.1 第一步确认作业是否真的用了去重优化先看执行计划。在 Flink SQL Client 里执行EXPLAIN SELECT user_id, last_login FROM ( SELECT user_id, last_login, ROW_NUMBER() OVER (PARTITION BY user_id ORDER BY last_login DESC) AS rn FROM user_login_log ) WHERE rn 1;结果里如果看到的是Rank节点而不是Deduplicate节点说明优化没生效。原因通常是last_login没有声明为时间属性。这一步能快速区分是优化没生效还是优化生效了但状态还是大。5.2 第二步检查状态大小和 key 基数如果计划里已经是Deduplicate但状态还是大那问题就在 key 的基数上。用 Flink Web UI 看Deduplicate算子的 State Size 指标同时估算一下 partition key 的基数。比如按user_id去重如果日活用户是 5000 万那状态里至少要存 5000 万条记录每条记录假设 200 字节就是 10GB 左右。这个量级单靠 TTL 是压不住的得从业务上想办法。5.3 第三步确认 TTL 是否配置且生效检查作业配置里有没有table.exec.state.ttl。注意这个参数是表级的可以在CREATE TABLE的WITH里配也可以在作业级别配。很多人以为配了就行其实还要确认 TTL 的时间单位——默认是毫秒如果你写table.exec.state.ttl 24那是 24 毫秒等于没配。正确写法是table.exec.state.ttl 8640000024 小时的毫秒数或者用带单位的写法取决于版本。5.4 第四步验证修复效果改完之后重新提交作业观察两个指标一是Deduplicate算子的 State Size 是否稳定在一个量级不再增长二是 Checkpoint 的 duration 是否恢复正常。如果 State Size 还在涨说明 TTL 没生效或者 key 基数确实太大需要回到第二步重新评估业务方案。整个排查链路的核心思路是先看计划再看状态最后看配置。顺序不能反因为如果计划本身就不对调配置是白费功夫。6. 那些文档里不会写的实操心得讲完了原理和方案最后分享几个我在实际项目里踩出来的经验这些在官方文档里基本找不到但能帮你少走很多弯路。6.1 ORDER BY 的字段选择直接决定状态大小同样是去重ORDER BY update_time DESC和ORDER BY amount DESC的状态开销可能差一个数量级。因为按时间排序时Flink 可以借助 watermark 清理旧状态按业务字段排序时它必须保留所有记录才能保证排名正确。所以能用时间字段排序就别用业务字段这是铁律。6.2 别在 ROW_NUMBER 外面套太多层子查询Flink 的优化器对嵌套子查询的优化能力有限尤其是多层嵌套里夹着ROW_NUMBER()很容易导致优化器放弃去重识别退化成普通Rank算子。我见过一个作业套了四层子查询结果状态是优化后的 20 倍。能合并的查询尽量合并实在要分层也要确保ROW_NUMBER()那一层是最内层。6.3 版本升级可能让原本能跑的 SQL 报错Flink 1.13 到 1.16 之间优化器对ROW_NUMBER()的校验越来越严格。有些在 1.13 能跑的 SQL升到 1.15 就报无法生成计划了。这不是 bug是校验规则收紧了。升级前一定要在测试环境把带窗口函数的 SQL 全部跑一遍别直接上生产。6.4 用 EXPLAIN 养成习惯每次写完带ROW_NUMBER()的 SQL提交前先EXPLAIN一下看计划里是Rank还是Deduplicate。这个习惯能帮你提前发现 90% 的问题比等作业挂了再排查高效得多。EXPLAIN 的输出里还能看到状态估算虽然不精确但能给你一个量级上的判断。6.5 状态 TTL 不是万能药很多人一遇到状态大就加 TTL但如果业务语义要求永久保留每个 key 的最新状态加 TTL 会导致结果错误——过期的 key 再来数据时会被当成新 key 处理排名就乱了。TTL 只适用于业务上能接受状态过期的场景用之前一定要和业务方确认清楚。7. 写在最后的一点个人体会ROW_NUMBER()在 Flink SQL 里的报错本质上不是语法问题而是流处理思维和批处理思维的碰撞。你带着批处理的直觉去写流 SQL就会觉得处处是坑一旦理解了无界、状态、时间属性这三个词的含义很多报错就变得理所当然了。我自己的习惯是写任何带窗口函数的 Flink SQL 之前先问自己三个问题这个排名的边界在哪状态会保留多少条记录时间属性声明了吗这三个问题想清楚基本就不会踩大坑。至于具体的语法细节查文档就行真正难的是把业务需求翻译成流处理能表达的语义。如果你现在正被某个ROW_NUMBER()报错卡住建议按这篇里的分类先定位属于哪一类再对照方案改。改完记得EXPLAIN验证别直接提交。踩过的坑多了你会发现这些报错其实都是 Flink 在提醒你你的 SQL 语义在流模式下不成立得换个思路。
阅读完成 · 觉得有帮助？
