Flink Table API 有界无界窗口实战：从语法到水位线避坑指南

之前在一个零售数据平台项目里，我经常要同时搞定两套统计逻辑：白天用实时任务统计每小时的订单流水，深夜还要把一整天的订单文件重新算一遍出日报。最开始的方案是流批两套代码分开维护，结果改一次指标口径要动两个地方，别提多折腾了。后来我把所有计算统一收敛到 Flink Table API，靠窗口把有界数据和无界数据两套逻辑拧成了几乎一样的写法。

这篇文章就把 Flink Table API 里“有界窗口”和“无界窗口”的使用彻底拆开讲清楚——包括窗口类型怎么选、时间属性和水位线怎么配、新旧窗口语法有什么区别，以及我实际踩过的一些坑。适合正在做实时数仓、流批一体开发，或者刚接触 Flink 窗口的同学，内容不算新手教程，但基本是能直接照着写的经验总结。

1. 先理清需求：有界与无界，窗口到底在解决什么问题

1.1 一条流的故事：有界数据和无界数据

很多人刚接触 Flink 时被“有界/无界”这个概念绕晕，其实一句话就能说清楚：有界数据流就是数据有限，读到最后肯定结束，比如文件；无界数据流就是数据一条接一条地来，不知道什么时候停，比如 Kafka 里的实时订单。

我习惯把这两类数据看成一本书和一条直播流。书有页数，翻到最后一页就完了，这就适合做批处理；直播没有终点，你想知道“开播 5 分钟内观众说了什么”，就得把这无限的内容切成一帧一帧的小段时间来观察——这就引出了窗口。

在 Flink Table API 里，这两个世界被统一成了同一种计算模型。文件可以当着有界流来读，Kafka 可以当着无界流来读，写完同一个窗口 SQL，引擎自己会判断怎么调度执行。这也是为什么说 Flink 是流批一体的计算引擎。

1.2 Table API 窗口的四种基本形态

Table API 中窗口并不是拍脑袋想出来的概念，它一共只有四种基本形态：滚动窗口、滑动窗口、会话窗口和累计窗口。四种窗口解决四类不同场景。

滚动窗口最好理解，比如 10 分钟的窗口，每 10 分钟算一次，窗口之间像切蛋糕一样互不重叠。滑动窗口麻烦一点，它会按照“步长”多次触发，同一个数据可能出现在多个窗口里，所以最后做聚合时经常会出现重复计算。会话窗口不看固定时间长度，而是看“间隔”，如果数据源头 5 分钟内没有新数据，那前面这段就封成一个窗口。累计窗口相对新一点，它只移动上边界，下边界固定，用于做“从今天 0 点累计到现在”这种统计。

1.3 为什么用 Table API 而不是直接写 DataStream

项目里我一开始用的是 DataStream API 加 ProcessWindowFunction，代码量大不说，改一个窗口逻辑要重新理半天数据流转换过程。换成 Table API 之后，最大的体感就是逻辑直接落在 SQL 语义上，窗口聚合写起来像写数据库查询，不用手动维护状态和定时器。

Table API 还自带优化器，比如做滑动窗口时，引擎可能会自动做算子复用；同样的逻辑用 DataStream 手动实现，还要纠结底层性能问题。而且 Table API 的窗口结果可以直接接到各种 Sink，比如 Kafka、ClickHouse、JDBC，甚至可以直接写入另一张动态表继续做下一层聚合。考虑到实时数仓里链路长、层级多，Table API 的开发效率优势极其明显。

2. 窗口语法的新旧版本差异与关键参数细节

2.1 老窗口语法与新 TVF 语法：改了什么

Flink 1.13 之前，Table API 窗口主要依赖 groupBy 里的分组窗口语法，代码长这样：

Table orders = tableEnv.from("orders"); Table result = orders .window(Tumble.over(lit(10).minutes()).on($"ts").as("w")) .groupBy($"item", $"w") .select($"item", $"w".start().as("window_start"), $"w".end().as("window_end"), $"amount".sum().as("total_amount"));

这就是老版本的分组窗口写法，把“窗口”作为分组 key 的一部分。这种写法有几个毛病：窗口和分组逻辑耦合在一起，后来想对窗口再做去重、TopN 这类计算时非常费劲。

1.13 开始官方主推窗口表值函数，也就是 TVF 写法，SQL 和 Table API 都支持。同样是 10 分钟滚动窗口：

SELECT window_start, window_end, item, SUM(amount) AS total_amount FROM TABLE(TUMBLE(TABLE orders, DESCRIPTOR(ts), INTERVAL '10' MINUTES)) GROUP BY window_start, window_end, item;

窗口变成了一个显式函数，先把数据按时间切分好，再交给后续聚合。这样做最大的价值是：你可以直接在窗口结果之后继续做基于窗口的 Join、TopN、去重等操作。我做实时大屏时，经常用这种写法先算窗口聚合，再对聚合结果做排名，SQL 写起来非常顺手。

新版本里四种窗口的 TVF 函数对应如下：

注意这里滑动窗口的第二个时间参数是步长，第三个是窗口大小，顺序别写反。我一开始就把 HOP 的步长和窗口大小搞反过，结果窗口计算结果都奇奇怪怪的。

2.2 三种时间属性：窗口能不能准确触发，全看这个

Table API 里窗口必须基于一个“时间属性”，你建表时指定的时间列，决定了窗口的语义。时间属性一共有三种。

处理时间最省事，比如用 DataGen 造数测试窗口逻辑时，我一般直接声明一列pt AS PROCTIME()，数据一进到算子就归到当前窗口，窗口立刻触发，非常适合验证计算逻辑正确性。

但真正生产环境，我几乎都会用事件时间。因为业务数据本身都有业务时间字段，比如订单的创建时间、日志产生时间。数据在网络上传输、缓冲之后，乱序是家常便饭，而事件时间配合 watermark 才能正确处理乱序问题。

事件时间在 Table API 建表时，声明方式通常是这样：

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

这里ts就是事件时间，WATERMARK表示允许最大 5 秒的乱序。意味着窗口不是一到时间就立刻关闭，而是等水位线越过窗口结束时间之后，才真正触发计算。这个细节非常重要，很多人窗口一直不出结果，基本都是 watermark 设置问题。

2.3 窗口大小、滑动步长与偏移量怎么选

窗口大小是最好定的一部分，业务上要求看多久就填多久。滑动步长决定了统计数据刷新的频率：比如大屏上要“每 5 分钟刷新一次最近 30 分钟的销售额”，窗口大小就是 30 分钟，步长就是 5 分钟。

偏移量这个参数容易被忽略。它解决的是跨时区问题。Flink 的滚动窗口默认为从1970-01-01 00:00:00开始按窗口大小切分。比如你想按自然日统计，但业务要求每天的零点在北京时间的 00:00:00，如果你的集群时区是 UTC，那默认切分出来的“每天”其实是 UTC 的零点，也就是北京的早上 8 点。

解决方式就是在窗口函数中加上偏移量。以 TVF 为例，滚动窗口的完整参数是四个：

TUMBLE(TABLE orders, DESCRIPTOR(ts), INTERVAL '1' DAY, INTERVAL '8' HOUR)

第四个参数就是时区偏移，东八区就加 8 小时。这个参数在新老语法里都有，老语法对应Tumble.over(...).withOffset(...)。项目里如果出现过“每天统计的数据对不上日期”这种问题，先查查是不是没设置偏移量。

3. 有界与无界窗口实操：从建表到跑通结果

3.1 环境准备与依赖

我用的 Flink 版本是 1.17，如果你用 1.14 之后的版本，下面这些代码基本通用。开发环境用 Maven 建工程，核心依赖就这几个：

<dependency> <groupId>org.apache.flink</groupId> <artifactId>flink-table-api-java-bridge</artifactId> <version>1.17.2</version> </dependency> <dependency> <groupId>org.apache.flink</groupId> <artifactId>flink-table-planner-loader</artifactId> <version>1.17.2</version> </dependency> <dependency> <groupId>org.apache.flink</groupId> <artifactId>flink-table-runtime</artifactId> <version>1.17.2</version> </dependency>

本地跑的话，再引入 flink-clients。工程里创建一个环境对象，这里统一用一个方法：

StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(); StreamTableEnvironment tEnv = StreamTableEnvironment.create(env);

后面所有实操代码，都基于这个tEnv对象往下写。

3.2 有界窗口实操：读取文件统计订单数据

有界数据最典型的来源是文件。我先准备一份 CSV 文件，里面存了一天的订单记录，字段包括用户 ID、商品 ID、金额和订单时间，格式如下：

1,1001,49.90,2024-06-01 10:15:00 2,1002,120.00,2024-06-01 10:25:00 1,1003,89.50,2024-06-01 10:40:00 3,1001,210.00,2024-06-01 10:55:00

用 Table API 的 filesystem connector 创建 source：

CREATE TABLE csv_orders ( user_id BIGINT, item_id BIGINT, amount DECIMAL(10, 2), ts TIMESTAMP(3), WATERMARK FOR ts AS ts - INTERVAL '5' SECONDS ) WITH ( 'connector' = 'filesystem', 'path' = '/tmp/orders.csv', 'format' = 'csv' );

注意文件数据在流模式下也是按照“有界流”来读的，文件读到最后会触发一个无限大的 watermark，所有没闭合的窗口都会一次性全部输出。这就是有界流的好处——你不用担心缺数据，只要文件完整，结果一定完整。

接着用 TVF 做滚动窗口聚合，统计每个商品每 10 分钟的销售额：

SELECT window_start, window_end, item_id, SUM(amount) AS total_amount FROM TABLE(TUMBLE(TABLE csv_orders, DESCRIPTOR(ts), INTERVAL '10' MINUTES)) GROUP BY window_start, window_end, item_id;

用 Table API 跑 SQL：

Table result = tEnv.sqlQuery(sql); result.execute().print();

实测下来，这份 4 条小样例最后会输出 2 个窗口的结果——10:10 到 10:20 的窗口包含第一笔订单，10:20 到 10:30 的窗口包含第二笔，假设数据量很小所以窗口序列不连续，其余窗口会被跳过，不会输出空白结果。这也是正确定义：不会输出空窗口。

有界场景最大的优势是可复现。同一份文件，无论你跑多少遍，结果一定一样，这非常合适做离线报表和测试数据链路。我在公司里做实时数仓时，经常用一份当天的 CSV 回放过去半天数据，用来验证新写的窗口逻辑是否正确。

3.3 无界窗口实操：DataGen 造数验证滑动窗口

无界数据我用 DataGen connector 来模拟，不用依赖 Kafka，本地就能跑起来。DataGen 会根据配置持续不断地生成数据，非常适合验证窗口的实时触发过程。

先建一张模拟订单表：

CREATE TABLE datagen_orders ( user_id INT, item_id INT, amount DECIMAL(10, 2), pt AS PROCTIME() ) WITH ( 'connector' = 'datagen', 'rows-per-second' = '5', 'fields.user_id.kind' = 'random', 'fields.user_id.min' = '1', 'fields.user_id.max' = '10', 'fields.item_id.kind' = 'sequence', 'fields.item_id.start' = '1', 'fields.item_id.end' = '100', 'fields.amount.kind' = 'random', 'fields.amount.min' = '1', 'fields.amount.max' = '1000' );

这里我为了演示无界窗口的实时触发，直接用了处理时间列pt AS PROCTIME()。事件时间无界流的建表方式跟 CSV 版本差不多，只要把 connector 换成 kafka 就行，逻辑可以复用。DataGen 每秒钟生成 5 条订单，配合处理时间，数据一进来窗口立刻结算。

接下来做一个滑动窗口统计，窗口大小 5 分钟，步长 1 分钟，相当于大屏上每一分钟都能看到过去 5 分钟每个商品的成交总额：

SELECT window_start, window_end, item_id, SUM(amount) AS total_amount FROM TABLE(HOP(TABLE datagen_orders, DESCRIPTOR(pt), INTERVAL '1' MINUTE, INTERVAL '5' MINUTES)) GROUP BY window_start, window_end, item_id;

跑起来之后，一分钟能看到一批增量输出。注意滑动窗口相邻结果之间的时间重叠，同一个数据可能出现在连续多个窗口中，这在语义上是正确的，统计“过去 5 分钟”本来就有重叠。

数据量大的时候，建议把rows-per-second适当调大，比如'1000'，这样窗口在短时间内就能看到比较明显的聚合变化。我一般造数测试不只跑一个窗口类型，会把滚动、滑动、会话全轮换跑一遍，确认计算逻辑在流模式下没有异常。

3.4 Table API 与 SQL 混合开发的两种写法

实操下来，Table API 有两种开发风格可以混着用。

第一种是纯 SQL 字符串。所有逻辑都写在字符串里，适合链路简单、一眼能看懂的聚合任务。比如刚才的无界滑动窗口，直接就是一行 SQL。缺点是字符串一长，里面字段名、窗口参数写错了不好排查，编译期发现不了。

第二种是 Table API 链式写法。它适合从某个 Table 对象出发，做一些动态拼接或者需要跟 DataStream 互操作的场景。比如：

Table orders = tEnv.from("datagen_orders"); Table result = orders.window(Slide.over(lit(5).minutes()) .every(lit(1).minutes()) .on($"pt") .as("w")) .groupBy($"item_id", $"w") .select($"item_id", $"w".start().as("window_start"), $"w".end().as("window_end"), $"amount".sum().as("total_amount")); result.execute().print();

这段就是个标准的滑动窗口 Table API 链式写法。老 API 和新 API 的区别在 2.1 提过，这里不再重复。

我个人的习惯是：窗口逻辑相对固定的，用 SQL 字符串；窗口参数需要根据外部配置动态变化的，用 Table API 链式写法。动态变化时，直接拼 SQL 很容易出错，比如传参没加引号导致时间单位解析失败，链式写法用 Java 方法参数更安全。

4. 常见问题与排查技巧：我踩过的坑都在这里

4.1 窗口不触发：水位线为什么迟迟不来

这是无界流里最典型的坑。窗口定义了，数据也源源不断进来了，但就是看不到输出。绝大多数原因都是 watermark 设置问题。

拿事件时间来说，watermark 代表“所有小于这个时间的事件都已经到达了”。如果数据里的时间戳离当前处理时间很远，或者很小概率出现的极端大时间戳把 watermark 一瞬间推到很远，窗口就会一直在内存里等后续数据，迟迟不触发。

排查方式我一般分三步：

1. 先看任务是否有一个 watermark 没有更新的迹象，可以通过 Flink Web UI 查看当前 watermark 数值；
2. 检查建表语句里WATERMARK FOR ts AS ...是否写对了，注意ts字段类型必须和实际数据格式匹配；
3. 检查数据源的时间字段是不是发生了异常，比如某个分区里的文件时间戳写到了未来时间。

有一个很隐蔽的问题：如果建表时没有声明WATERMARK，Flink 会把事件时间当普通字段，窗口在事件时间语义下是不会正确触发的，它根本不知道水位线是什么。这个错误我见过不少人犯。

4.2 时间字段报错：TIMESTAMP 类型为什么总不对

Table API 对时间字段类型很严格。比如把ts定义为TIMESTAMP(3)，但你 CSV 里写的是2024-06-01 10:15:00.123456，毫秒精度对不上，解析阶段就可能报错。

另一个高频报错是TIMESTAMP_LTZ和TIMESTAMP混淆。带时区语义的时间，在事件时间和 water mark 计算中建议使用TIMESTAMP_LTZ(3)；TIMESTAMP(3)是本地时间戳不带时区信息。如果数据里带Z后缀的 UTC 时间，要用TIMESTAMP_LTZ。搞混后窗口计算出来差 8 小时，这是很经典的问题。

建议所有时间字段统一规范化：源头数据在清洗阶段转成毫秒时间戳，建表时统一用TIMESTAMP_LTZ(3)，展示层需要读成可读时间时再用DATE_FORMAT之类的函数转。这套规范在很多项目验证过，可以省很多排查时间。

4.3 有界流的一个冷门陷阱：拿到全量才能出结果

有界文件流虽然本质上是流，但仍然不是“来一条算一条”。窗口没有实时触发的原因不是数据处理得慢，而是有界流的水位线要到文件读完了才会变成正无穷大，所以在流模式下读取文件的窗口，通常要等整个文件读完，才会把结果刷出来。

很多人第一次用 filesystem connector 跑窗口时，容易产生“为什么等了很久没结果”的疑问。这不是 bug，而是有界流的语义决定的：只有在保证不会有新数据进来之后，窗口才能安心关闭。理解这一点后，你在做离线文件报表时，就不要指望它能像实时流一样边读边出结果。要验证实时触发，必须用无界数据源。

4.4 迟到数据怎么办：窗口边界和侧输出

事件时间窗口即使有 watermark，也可能遇到极度晚到的数据。Table API 提供了两种处理策略：一是容忍乱序，watermark 本身就是在告诉引擎我能接受多大延迟；二是设置allowedLateness，窗口关闭后，还会再等一段时间允许数据修改结果。

在老 Table API 里，可以通过.allowedLateness(lit(1).minutes())设置；新 TVF 写法中，会在查询后面声明：

SELECT ... FROM TABLE(TUMBLE(TABLE orders, DESCRIPTOR(ts), INTERVAL '10' MINUTES)) GROUP BY window_start, window_end, item;

配合lag或者直接设置表参数来控制。如果你的业务允许结果晚几分钟出，加 allowed lateness 是性价比最高的方案。

但注意，迟到数据在窗口重新计算时，可能导致下游已经输出的指标发生修正。比如已经算好发出去的报表，又被一条迟到数据拉高了数值。像这种场景，我建议在架构上把“首次输出”和“最终修正”分成两条链路，或者在下游接受基于最后到达时间的修正。这也是流计算里最需要业务设计共同决定的事。

回到开头那个项目，最后我们只维护了一套 Table API 窗口代码，白天跑无界 Kafka 流，深夜把同一套逻辑换一个有界文件 Source 再跑一遍，两边结果基本能对上。窗口从表面上看是个技术概念，实际上它是把无限数据切成有限块反馈给业务的通用工具。

我个人实操中还有个小习惯：新写窗口逻辑时，先拿 DataGen 模拟无界流把实时触发的正确性验证掉，再拿一段小文件做有界验证，两边口径一致了才接真实 Kafka 数据。这个流程帮我避免了很多“上线即出 bug”的尴尬。后续如果你的业务还需要做窗口内的 TopN、去重，或者多流窗口 Join，完全可以基于这套 TVF 语法继续扩展，Flink Table API 在这块的表达能力足够应付大多数场景。