Flink SQL 1.13窗口函数重构与状态调优实战指南

我对 Flink SQL 1.13 的印象，不是从发布公告里看出来的，而是从一次凌晨的报表事故里印出来的。那时候我负责的实时任务还在用 1.12 的旧窗口语法，按小时窗口聚合订单量。某一天观察窗口，某个时段的产出全部为零，拓扑看起来又是健康的，没有反压，数据水位也正常。后来把 Flink SQL 升级到 1.13，用新的窗口表值函数（Table-valued Function，TVF）重写了查询，才真正理解了问题所在：旧窗口语法把窗口当作分组里的特殊函数来理解，语义和优化方式都很绕，出了问题不容易一眼定位。

这篇文章我不会把官方特性列表再抄一遍，而是按照一个普通工程师在实际项目里会遇到的问题顺序来写：窗口、水位、状态、维表关联、版本迁移、内存和检查点。希望你看完以后，不只是知道 Flink SQL 1.13 有哪些新东西，更能知道这些新东西到底该用在哪、怎么用、以及上线前要盯住哪些指标。

1. 窗口表值函数的重构：从“按时间函数聚合”到“把窗口当表来用”

1.1 旧窗口与 TVF 窗口的 SQL 差异

在 1.12 及之前，Flink SQL 写滚动窗口的常规姿势是这样的：

SELECT TUMBLE_START(ts, INTERVAL '1' HOUR) AS start_time, TUMBLE_END(ts, INTERVAL '1' HOUR) AS end_time, COUNT(*) AS order_cnt FROM orders GROUP BY TUMBLE(ts, INTERVAL '1' HOUR);

这段代码本身没有毛病，跑一轮小时窗口聚合问题不大。但是只要想继续做“窗口套窗口”，比如先按 5 分钟计算一次，再把 5 分钟结果汇成小时窗口，旧语法就很难写。你只能通过TUMBLE_START算出来的时间字段再去嵌套 GROUP BY，而且还不能直接用函数结果作为新的时间属性，得手动对齐边界。换句话说，旧窗口是优化器里的“特判逻辑”，不是普遍规范的一等公民。

1.13 里，窗口被重新设计成表值函数，日常写法变成了下面这样：

SELECT window_start, window_end, COUNT(*) AS order_cnt FROM TABLE( TUMBLE(TABLE orders, DESCRIPTOR(ts), INTERVAL '1' HOUR) ) GROUP BY window_start, window_end;

TUMBLE不再放在 GROUP BY 里，而是放在 FROM 子句里，以TABLE(...)的形式把输入表“切”成一个带window_start、window_end两列属性的虚拟表。你要做的就是按这两列做普通 GROUP BY，窗口信息和普通字段在语义上完全对齐。

这种变化给开发带来的直接影响：结果集里天然带窗口开始与结束时间，不需要额外调窗口辅助函数；窗口属性作为普通列存在，可以继续做 JOIN、UNION、再分组；优化器能把窗口 TVF 的表结构纳入统计和算子合并，而不是把窗口特殊处理；后续做级联窗口也简单了，只要把前一层窗口的window_end作为下一层窗口的时间列继续传下去。

执行计划层面，旧的窗口聚合需要“窗口函数 + 分组键”协同工作，状态划分粒度比较粗；新 TVF 让窗口产出的窗口边界成为普通字段，状态可以跨窗口共享，并且便于做 pre-aggregation。这里面最大的价值不是性能一定更好，而是语义和优化路径清楚了，出了问题你知道去哪里看。

1.2 窗口偏移、时区与左闭右开边界

我这次踩的坑恰恰和边界对齐有关。业务是按照自然日统计，但订单产生的“日期”要从上午八点才能开始算一天，也就是说业务日期跨过了 UTC 零点。旧语法的窗口函数默认从开始时间截断到周期零点，要按业务边界对齐，得手动去算偏移。1.13 的滚动窗口 TVF 支持第二个时间间隔参数作为偏移量，比如这样写：

SELECT window_start, window_end, COUNT(*) FROM TABLE( TUMBLE(TABLE orders, DESCRIPTOR(ts), INTERVAL '1' DAY, INTERVAL '8' HOUR) ) GROUP BY window_start, window_end;

注意window_start和window_end是左闭右开区间，即[window_start, window_end)。对于按小时窗口聚合，发生在整点那一瞬间的数据归前一个小时。偏移量和这个左闭右开的关系需要小心，如果手动把窗口起点往前挪一小时，你实际上会看到两个小时的边界出现二十分钟的重叠。很多第一次用偏移参数的同事会在这上面花半天时间。

代码中需要确保ts在 DDL 中的类型是TIMESTAMP(3)或者TIMESTAMP_LTZ(3)，如果源字段是毫秒级BIGINT时间戳，就得先转成 TIMESTAMP 再传给 DESCRIPTOR。时间列的精度也会影响窗口对齐，毫秒和秒级在处理边界的时候差异不大，但如果你把 watermark 设置得太靠后，边界数据很晚才被触发，就会表现为一个窗口彻底不出结果。

最后再提一下时区。1.13 中有一个全局参数table.local-time-zone，影响CURRENT_TIMESTAMP、PROCTIME()以及窗口函数的时区转换。如果你只在 JVM 启动参数里改了时区，但 SQL 端没有同步，很可能出现窗口边界和本地时间对不上的怪问题。上线之前最好把 SQL Client 和作业参数里的时区统一。

补充一句：升级到 1.13 之后，旧的TUMBLE()/HOP()/SESSION()在 GROUP BY 里的写法仍然能用，但会在日志里给出弃用提示。我建议在近期排期里全部迁到 TVF 写法，尤其如果你还准备继续升到 1.14/1.15，尽早适配成本最低。

2. 水位控制与延迟数据的边界：事件时间窗口怎么才算“真的结束”

2.1 事件时间列的声明和水位的生成

Flink SQL 1.13 里事件时间窗口能不能正常触发，前提是源 DDL 里WATERMARK定义正确。

在流式 DDL 里，常见的写法是：

CREATE TABLE orders ( order_id BIGINT, item_id BIGINT, ts TIMESTAMP(3), WATERMARK FOR ts AS ts - INTERVAL '5' SECOND ) WITH ( 'connector' = 'kafka', ... );

这里的意思是：事件时间列是我们收到的业务时间 ts，允许最多 5 秒乱序。引擎按这个表达式推导出水位，不管实际数据到没到，只要水位越过了窗口的window_end，窗口就会关闭并输出结果。

这个版本里，时间属性已经绑定到列上，入门阶段错误少了很多。之前开发总会把时间字段命名为event_time或rowtime，实际在 DDL 里绑定了没关系的列，结果水位出来得比合理值更早或更晚。建议遵循一个原则：时间属性在 DDL 中只声明一次，后续所有窗口和 join 都用同一个字段，逻辑就会清楚很多。

2.2 空闲源和水位的自动推进，延迟数据怎么办

我曾经在双流关联的场景里遇到过一个特别典型的问题：左流数据完整，右流数据量稀稀拉拉，如果右流水位长时间不前进，JOIN 的窗口就会迟迟不触发。传统做法是设置table.exec.source.idle-timeout：

table.exec.source.idle-timeout: 10s

这样左流在右流空闲超过 10 秒时，会自动推进水位，让窗口继续计算，而不是被空闲源卡死。这个参数非常值得写进从 1.12 升到 1.13 时的排查清单，因为它往往不是你主动去配置的，而是当你发现“所有窗口都不产出了”以后才回忆起来。

至于迟到数据，普通 SQL 窗口聚合默认的策略是：迟到的记录如果已经落在水位之下，窗口已经关闭，就丢弃。1.13 的 SQL 层没有类似 DataStream APIallowedLateness那种随手配置的开关，你要是真要做延迟修正，一般是在 Sink 端做幂等回撤，或者用 TopN 去重重新计算。因此我强烈建议：上游源头的时间戳规范必须做好，宁可给一个稳定的乱序容忍（比如 30 秒），也不要为了追求“低延迟”把容忍压得太紧，否则晚到的重试消息真的很难处理。

水位问题的另一个常见表现是，窗口一直不关闭，查了半天发现是 Kafka 分区没有数据，但任务没有进入 idle 状态。此时需要确认 idle-timeout 是否覆盖到所有 Source 算子。还要注意上下游多个并行子任务共享同一个时间属性时候，watermark 取的是所有输入分区的最小值，只要有一个分区稳态为空，就得靠 idle-timeout 来解。

3. 去重与分组聚合的状态膨胀：一次从内存超限到定位的完整排查

3.1 基于 ROW_NUMBER 的流式去重到底存了什么

1.13 在流式 SQL 里去重，常见方式是用ROW_NUMBER()加上WHERE rn = 1：

SELECT order_id, pay_time, amount FROM ( SELECT *, ROW_NUMBER() OVER (PARTITION BY order_id ORDER BY event_time DESC) AS rn FROM payments ) WHERE rn = 1;

这个语法简单，但和离线 Hive 去重有本质差别：Flink 需要把所有分区键对应的当前最优记录都保存在状态里，因为流是不停的，你不知道下一个事件会不会比当前最新的还要新。换句话说，PARTITION BY order_id构成了状态的主键，状态记录数等于去重键的实时基数，而不是“只留最终结果”那么少。

我维护的一个支付去重任务，平常状态量稳定在 2GB 左右，结果某一天一个大客户跑批量补单，一个 order_id 反复出现不同状态的流水。此时状态后端是 RocksDB，磁盘占用和恢复时间同时上升。一开始我们只给 taskmanager 加内存，其实没有解决问题，因为 RocksDB 的列并不完全占 JVM 堆。

排查链路大致如下：

1. 先看 TaskManager 里的状态指标，确认Current number of entries in state是否比前一天明显放大。
2. 再对比对应 key 的数据形态，发现补单场景产生了不同pay_time的重复记录，而我们的去重排序字段是event_time，每次补单都会生成一条新的时间戳。
3. 把排序字段从“接口重试时间”换成“源系统里的支付完成时间”，并把补单标志作为过滤条件，重复数据量立刻下降。
4. 明确各类 key 的合理生命周期后，给状态设置 TTL，从根上控制无界增长。

这里的关键不是换一个“性能更好的去重语法”，而是搞清楚去重键的基数到底有多大、状态里每个键保留多少版本的数据。

3.2 DISTINCT 聚合拆分与 TTL 参数的实际效果

如果任务是统计某个维度下的COUNT(DISTINCT user_id)或COUNT(DISTINCT order_id)，这种聚合天然会保留全量键，状态压力比普通统计大得多。1.13 提供了拆分优化开关：

table.optimizer.distinct-aggregate.split.enabled: true

开启后，优化器会把一个COUNT(DISTINCT a)的聚合拆成“局部去重 + 全局去重”两层，让不同的上游并行子任务在自己的分区内先做一次去重，再交给全局聚合节点。对于键基数较大的场景，局部聚合能显著减少向上传递的重复记录数。

我这里有过一组参考数据：不开启时，5 个并行分区产生的中间记录大约每秒 20 万条；开启后，中间记录下降约 30%。注意不同 key 分布效果不同，需要实际测试；如果本身 key 的分散度很平均，并且去重键基数不大，提升空间就要保守看待。

TTL 是另一个关键参数，可以这样设置：

table.exec.state.ttl: 1 h

这个参数直接作用于 SQL 状态中的过期数据。流式去重或普通聚合状态如果没有设置 TTL，RocksDB 会一直保留所有 key 的最新值。时间一长，作业就像长期不清理缓存的应用，内存越占越大，恢复也越来越慢。

但 TTL 不能拍脑袋设。如果业务上需要对 7 天内的订单做严格去重，把 TTL 设为 1 小时就会导致重复数据漏掉。我建议先盘点业务需要的最大回看窗口，在这个期限上加一些余量。Flink 的 TTL 是惰性删除，过期数据要等访问或者后台清理才会被淘汰，所以不要把它当成精确的数据生命周期管理工具，它更像是一道安全边界。

4. 维表关联的缓存策略：正确性和延迟之间的妥协

4.1 Lookup Join 在 1.13 的正确打开方式

流上来以后用订单中的地点 ID 去关联一份实时维表，1.13 里标准 SQL 写作：

SELECT order_id, l.region_name FROM orders AS o LEFT JOIN dim_location FOR SYSTEM_TIME AS OF o.proc_time AS l ON o.location_id = l.location_id;

这里FOR SYSTEM_TIME AS OF o.proc_time是时间语义，意味着关联时取 l 表中“当时”的最新版本。由于proc_time是处理时间，查询在每条记录刚到达时就发维表请求，所以本质上是一种以事件驱动、查询外部系统的 lookup join。

在 1.13 版本，维表关联性能的关键不在 SQL 本身，而在 connector 提供的 lookup 缓存和并发能力。如果每次都直接查外部系统，维表接口 QPS 很容易变成瓶颈。很多项目里都会这样配置：

WITH ( 'connector' = 'jdbc', 'url' = 'jdbc:...', 'lookup.cache' = 'PARTITION', 'lookup.cache.ttl' = '5 min', 'lookup.cache.max-rows' = '10000' )

lookup.cache表示缓存粒度，PARTITION意思是每个并发分区维护自己的缓存，这样不会互相抢占 key；lookup.cache.max-rows是最大缓存条数，超过后按照清理策略淘汰。

4.2 缓存参数怎么调，以及外部数据变更的影响

这里最容易被低估的问题：维表数据在 SQL 执行期间发生变化，缓存 TTL 决定你能得到多“新”的维度。

如果业务允许维度变更在 5 分钟内生效，TTL 可以设置成 300 秒，外部系统压力最小。如果订单金额直接影响结算，希望维度变更立刻生效，那就只能把 TTL 调小到秒级，但此时外部存储的 QPS 会陡增，甚至有些关系型数据库维表会被任务的并发查询拖垮。

我实际采用的折中办法是分场景：

• 高频流量类场景：大促活动期间，流量突然放大，先把 TTL 从 5 分钟改成 1 分钟，同时让维表服务落到独立只读副本，给主库减压。
• 离线或内部统计分析：TLL 设置 15 分钟以上，正确性指标不再敏感，尽量保护外部系统。

另外一个常被忽略的参数是 lookup 的并行度。如果维表关联算子并行度设置太低，数据流到了这个算子之后会排队，表现为整个任务的反压点在Calc/Lookup Join附近。可以给维表关联算子单独设置并行度，或者直接把整个 SQL 并行度抬上来，看外部系统能否承受。如果你发现 CPU 不高、外部数据库延迟也很低，但反压还在这里，大概率是并行度不够。

还有一点：lookup join 在 SQL 中的默认语义是关联失败返回 null，并不会阻塞主流程。所以在业务上要对 null 结果做兜底，比如用COALESCE(l.region_name, '未知区域')，甚至后续用一个异步补数算子把没关联上的记录重新查一遍。既然做了重试，就必须把重试的幂等性设计好，否则同一个订单可能生成两条最终结果。

5. 从 1.12 升到 1.13，容易被忽略的四个兼容点

5.1 语法弃用只是表层警告

升级之后最不用慌的是旧窗口 GROUP BY 的弃用提示。任务还能跑，只是警告。真正要留意的是那些没有告警、但执行计划变化了的 SQL。比如某些内置时间函数在不同版本中对时间精度的舍入规则有细微变化，如果你的业务里正好用到分钟级别的窗口切片，就需要把结果直接对拍一遍。

做法很简单：拿一段固定输入，分别在 1.12 和 1.13 SQL Client 里跑同一条查询，对比输出。尤其关注window_start和window_end这类的边界字段，以及精确到毫秒的时间字段。差异不一定是 bug，但要用一张对照表记录差异，确认是否符合业务预期。

EXPLAIN是升级过程中最实用的工具。升级以后先看EXPLAIN PLAN FOR <your_sql>，对比前后执行计划里的算子变化。如果某一版出现了你没见过的TableSourceScan或者ChangelogNormalize算子，就要关心它到底在做什么、会不会引入额外的状态与开销。

5.2 配置项与运行参数的不兼容

很多老任务从 1.12 带着一份 flink-conf.yaml 直接升级到 1.13，这种习惯在高版本升级里很危险。1.13 对部分配置项的命名和默认值有调整，尤其是内存参数。你可能在旧配置文件里写了taskmanager.memory.size这类曾经有用的键，新版本会强调taskmanager.memory.process.size或者taskmanager.memory.flink.size来控制最终进程大小。

更值得注意的还有状态后端配置。从 1.12 直接升上来，最好显式确认：

• state.backend取值是rocksdb还是filesystem；
• state.backend.incremental是否开启；
• state.backend.rocksdb.memory.managed是否开启。

升级后第一次启动的状态初始化会把这些配置重新读一遍，如果有参数没有识别，日志里会打出 unknown config key。别只看第一条报错，把日志里所有 unknown key 收集起来一条一条对照官方文档，而不是直接“先跑起来再说”。跑起来容易，等内存超限再查就难了。

5.3 checkpoint 状态恢复的验证方式

状态兼容性没法靠文档拍胸脯，必须实测。我升级的流程是：

1. 老版本正常跑，手动创建一个 checkpoint，保存 checkpoint 路径和元数据。
2. 用 1.13 的作业，在启动命令里指定-s <old_checkpoint_path>。
3. 让作业空跑 10 分钟，观察 checkpoint 是否成功、状态指标是否异常反弹。
4. 然后把真实数据源接进来，对比升级前后的历史结果和当前输出。

如果你启用过非对齐 checkpoint，或者用到了状态 TTL 外的自定义状态，兼容性更要仔细验证。如果恢复失败，不要第一时间怀疑是新版本的 bug，先用缩小版状态集去试，排除状态数据损坏的因素。

6. 内存模型与检查点调优：让长时间运行的任务少“抖动”

6.1 内存配置的拆解思路

从 1.13 开始大家都在讲“内存模型”四个字，实际上真正需要理解的是 taskmanager 内存被分成了几块：

• JVM 堆内存，包含框架内存和 Task 内存；
• 托管内存，用于 RocksDB、排序缓冲等；
• 网络缓冲内存；
• JVM 直接内存和元空间。

你在容器里看到进程占用的内存，不等于你顺手调大堆大小就够了。很多时候任务频繁 GC，其实堆内存设置完全合理，是 RocksDB 的托管内存太小，或者网络缓冲被某个窗口数据量压爆。

一些开发会直接设置taskmanager.memory.process.size为一个固定字节数，比如 8GB，然后让 Flink 根据默认比例去分配。这个思路可行，但前提是你知道不同模块的默认子比例。如果你的状态量大、窗口聚合多，建议显式调大托管内存：

taskmanager.memory.managed.fraction: 0.8 taskmanager.memory.managed.size: 4096mb

同时打开 RocksDB 的内存托管：

state.backend.rocksdb.memory.managed: true

让 RocksDB 尽量使用托管内存而不是随意抓取 JVM 堆。对于跑在容器调度平台上的任务，还要记得为 JVM 元空间和本地磁盘缓存预留空间，否则进程会被系统 OOM Killer 杀掉。

6.2 非对齐检查点的收益和代价

流式计算的分布式快照长期以来要求所有算子对齐 barrier，一旦某个 task 反压，barrier 就会排队，于是 checkpoint 间隔被拉长。1.13 对这类场景提供了更实用的非对齐检查点配置。

示例：

execution.checkpointing.interval: 60s execution.checkpointing.alignment-timeout: 30s execution.checkpointing.timeout: 10min

当对齐时间超过alignment-timeout，引擎会放弃对齐，直接生成非对齐 checkpoint。收益是反压期间也能产出检查点，不会因为长时间没有检查点导致故障恢复时丢失过多数据。代价是传输中的记录也会被写入检查点，也就是网络缓冲和未处理消息会占用额外空间。

我的建议是：默认不打开非对齐，只在反压常态化的长链路任务上开启。开启后监控磁盘上的 checkpoint 文件大小，如果比对齐模式大了好几倍，说明链路堆积严重，要么加并行度，要么优化瓶颈算子，而不是靠不停拍快照兜底。

恢复速度方面，1.13 支持state.backend.local-recovery: true，让同一个 taskmanager 下次恢复时优先读本地状态，减少跨网络拉取状态的时间。这个对大规模 RocksDB 作业很管用，但要注意如果本地磁盘坏了，还是会从远程恢复，所以本地盘稳定性需要评估。

7. 上线前我固定跑的一套验证流程

7.1 小压强下的数据回放测试

升级 1.13 之前，我一定会先在测试环境里跑一次缩小版数据回放。把生产环境最近 2 小时的一小段数据从源端 dump 出来，按照原来的 event time 顺序发到测试 topic，再把同一套 SQL 分别跑在 1.12 和 1.13 上，最后对比结果表输出。这时要做的事情不只是对比数值，还要对比窗口边界、不同时区下的边界、window_start精确到毫秒的格式。

如果 SQL 里用到了去重，就用一组构造相同的主键在系统不同时间点循环插入，观察结果表中是否出现重复。如果 SQL 里有维表关联，测试环境中维表的初始数据和生产库保持一致，再用小批量流去刷新缓存，看 TTL 到期后结果是否变化。很多问题第一轮部署到测试环境里就能跑出来，不要等到大促前才动。

7.2 监控哪些指标算“跑稳了”

任务在测试环境跑起来后，我会盯这几个指标：

• SourceOperator.WatermarkLag：当前处理时间与水位之间的差距是否稳定，差距太大说明流里的时间戳分布或乱序配置有问题。
• Task 级别的状态大小：是否持续上升不回落。如果是单调递增，说明 TTL 没有生效，或者业务维度过宽。
• Checkpoint 耗时与失败次数：新版本升级后容易出现第一次 checkpoint 卡住不动，背后往往是状态兼容问题。
• 反压比率：如果长时间大于 0.5，先别急着优化 SQL，先看维表关联是不是把外部系统打满了。

这些指标是所有版本通用的，但 1.13 升级后更容易被执行计划变化放大。比如某个之前不存在的ChangelogNormalize算子引入了新状态，你如果不看指标，根本不知道状态已经开始涨。所以我更喜欢把“看计划、看状态、看水位”三件事放在同一次上线验证里做，能省掉很多来回沟通的成本。

Flink SQL 1.13 对 SQL 来说是一次语义收敛。窗口变成表值函数之后，很多旧语法时代的“特例”都被规范化了，代价是执行计划比以前复杂，陌生算子也更多。深入理解一个版本，不应该只是背下来几条语法，而是把语法、状态、内存、恢复串成一条心里有数的链路。我在实际使用中最深的一点体会是：版本升级的成功率，往往不取决于新特性多漂亮，而是取决于你对老任务每个算子的状态和时间理解有多深。多跑一份 EXPLAIN，多打一次 checkpoint，多盯一次水位，很多事故在发生前就已经在指标里露头了。