做实时计算这几年“窗口”这个词几乎每天都绕不开。数据流本身是无限的但业务问题通常是有边界的我要看每分钟的PV、每小时的订单额、最近10条数据的平均时长这些都需要在无限流上切出有限的一段。Flink的Table API和SQL把窗口能力做成了声明式的官方把窗口明确分成两大类Group Windows分组窗口和Over WindowsOver窗口。这篇笔记是我系列笔记里的第16篇主要把我在实际项目里用Flink Table窗口的总结整理出来包括两类窗口分别怎么用、和DataStream API的窗口有什么差别、Event Time/Processing Time字段怎么配合以及哪些坑我踩过。适合已经把窗口从DataStream API搬到SQL的开发者也适合正在用Table API写窗口但偶尔报错、搞不清语义的人读完可以直接照着示例改。1. 窗口的本质为什么Table API要分Group Windows和Over Windows1.1 流计算里“无限”和“有限”的矛盾流计算的数据是持续到达的理论上没有终点所以任何聚合都不能直接在整条流上做。窗口做的事情很简单把无限的数据流按时间或行数切成有限的分片每个分片内部做聚合然后随着时间推进不断输出结果。你可以把窗口理解成一把“尺子”每隔一段量一次。但同样是切数据业务需求有两种完全不同的形态。第一种是“把一堆数据凑成一整块再算”例如统计过去5分钟的总成交额算完把这块数据“消费”掉输出一条结果。第二种是“每一行经过时都能看到以它为中心的一个邻域”例如每条订单进来我都想知道这单之前的10单平均价格或者最近1小时累计成交额输出结果时每一行都保留同时附上这个邻域里算出来的值。Flink Table API和SQL分别用Group Windows和Over Windows来对应这两类需求。1.2 Group Windows和Over Windows的分工Group Windows就是大家更熟悉的滚动窗口、滑动窗口、会话窗口。它的特点是窗口内数据先被收集再整体聚合每个窗口或每个窗口内每个分组最终输出一条结果窗口之间有明确的划分和边界。Over Windows则来自SQL标准里的OVER子句它不会把数据“切成块”而是为每一行计算一个基于它之前或之后若干行的聚合值输出行数与输入行数保持一致。两类窗口适用场景完全不同我整理了一个对比写代码前先想清楚自己到底要哪种对比维度Group WindowsOver Windows输出行为每个窗口/分组输出一条聚合结果每一行都输出附带聚合值数据是否保留窗口结束即关闭块内数据不保留所有原始行都保留逐行计算典型场景离线化统计分钟级PV、小时级GMV移动累计最近N条均值、累计金额SQL对应GROUP BY 窗口函数OVER 聚合函数窗口类型Tumble / Slide / SessionROWS / RANGE 两种边界对时间字段的要求聚合前切分窗口字段可参与GROUP BY必须ORDER BY时间字段否则没意义这个表基本就是你选择窗口方案的判断依据。想统计时间段内的总量用Group想看每一行的移动趋势用Over。两者没有谁替代谁的问题实际项目里经常组合使用。1.3 从DataStream API迁移到Table API窗口到底变了什么如果你之前写过DataStream API可能会觉得Table API的窗口“长得不太一样”。DataStream里先keyBy再timeWindow然后写apply或process函数Table API里变成了一段SQL或链式表达式声明窗口、声明分组字段、声明聚合函数剩下的交给优化器。这是从命令式到声明式的转变命令式写“怎么实现”声明式写“要什么结果”。Table API的窗口不直接操作DataStream而是作用在“动态表”上。动态表会随时间不断变化窗口聚合的输出本身也是一张动态表它会在窗口结束时追加新行或者在结果发生变化时更新旧行。很多人在这个点绕不过去为什么一个GROUP BY TUMBLE(...)在流模式下会有更新语义因为只要分组键不是时间本身同一个分组可能在不同窗口重复出现Flink需要区分这是追加还是更新。这一点在选Sink的时候特别重要后面的实操案例里我会专门提。2. Group Windows实战滚动、滑动、会话窗口的正确写法2.1 Tumble滚动窗口每N时间切一刀滚动窗口是最简单的窗口窗口之间不重叠每条数据只会属于一个窗口。比如“每分钟统计一次访问量”就是典型的Tumble窗口。Table API里用TUMBLE(time_attr, INTERVAL 1 MINUTE)来声明其中第一个参数必须是时间字段第二个参数是窗口长度。SQL写法SELECT TUMBLE_START(ts, INTERVAL 1 MINUTE) AS win_start, COUNT(*) AS pv FROM user_actions GROUP BY TUMBLE(ts, INTERVAL 1 MINUTE);Table API DSL写法Java 1.13风格示意Table result tableEnv .sqlQuery( SELECT TUMBLE_START(ts, INTERVAL 1 MINUTE) AS win_start, COUNT(*) AS pv FROM user_actions GROUP BY TUMBLE(ts, INTERVAL 1 MINUTE));这里最关键的是窗口辅助函数。TUMBLE_START返回窗口开始时间TUMBLE_END返回窗口结束时间注意是开区间末尾TUMBLE_ROWTIME和TUMBLE_PROCTIME分别返回可下推的Rowtime和Proctime属性可用于后续窗口连接或生成时间列。很多人问为什么一定要GROUP BY TUMBLE(ts, ...)而不能只GROUP BY TUMBLE_START(ts, ...)因为窗口本身才是分组键辅助函数只是从分组键里取属性出来。2.2 Slide滑动窗口固定长度加固定滑动步长滑动窗口用来做“每5分钟看过去1小时”这类重叠统计。窗口长度和滑动步长可以不一样数据可能同时属于多个窗口。声明方式比Tumble多一个参数HOP(time_attr, slide_interval, size_interval)第一个是滑动步长第二个是窗口长度。注意顺序习惯上容易写反。SQL写法SELECT HOP_START(ts, INTERVAL 5 MINUTE, INTERVAL 1 HOUR) AS win_start, HOP_END(ts, INTERVAL 5 MINUTE, INTERVAL 1 HOUR) AS win_end, COUNT(DISTINCT user_id) AS uv FROM user_actions GROUP BY HOP(ts, INTERVAL 5 MINUTE, INTERVAL 1 HOUR);滑动窗口的语义需要特别注意不是先按长度切好再等步长触发而是按步长生成一系列起点窗口终点的确定方式是按步长推进的。实际效果就是每5分钟会输出一次结果每次结果覆盖过去1小时的数据。窗口重叠意味着同一条数据可能被计入多个窗口这是业务想要的不是Bug。2.3 Session会话窗口按活跃间隔切分会话窗口不依赖固定长度而是依赖“活跃间隔”。连续两条数据的时间差超过指定GAP就认为上一个会话结束、新会话开始。它特别适合做用户连续行为分析、在线状态划分。SQL写法SELECT SESSION_START(ts, INTERVAL 10 MINUTE) AS session_start, SESSION_END(ts, INTERVAL 10 MINUTE) AS session_end, user_id, COUNT(*) AS action_cnt FROM user_actions GROUP BY SESSION(ts, INTERVAL 10 MINUTE), user_id;会话窗口在事件时间下比较“挑剔”它需要更精确的状态管理因为会话边界要到数据不再出现才能确定所以延迟数据对会话窗口的影响比滚动窗口更大。实际项目里如果数据迟到乱序严重会话窗口的结果容易出现“会话被拆开”或者“迟迟不能关闭”的现象需要配合合理的水位线和空闲分区处理来缓解。2.4 分组窗口的几个容易忽略的语义细节分组窗口虽然好写但语义细节有几个容易踩。第一GROUP BY里必须包含窗口字段这是语法强制的。如果SELECT里只写了TUMBLE_START(...)而GROUP BY里只有普通分组字段Flink会直接报错。如果你只想按窗口聚合、不需要其他分组字段那GROUP BY就只写窗口函数。第二窗口辅助函数里的参数是时间字段不是窗口别名。标准SQL里有些数据库支持GROUP BY TUMBLE(...) AS t然后SELECT t.start但Flink的语法习惯是直接TUMBLE_START(ts, ...)别混着用。第三窗口时间的开闭区间。Flink的窗口和DataStream一样左闭右开。[win_start, win_end)这也是为什么TUMBLE_END看起来是“下一秒/下一分钟”而不是窗口最后一行的时间。第四聚合函数的选择。窗口内支持COUNT、SUM、AVG、MAX、MIN、DISTINCT聚合在SQL里可以直接写。如果是Table API表达式写法也支持但COUNT(DISTINCT field)在流模式下状态开销会明显大很多做UV类指标时要做好状态清理的心理准备。3. Over Windows实战ROWS与RANGE边界怎么选3.1 Over窗口到底在算什么Over窗口不会重排你的输出结构它只是在你每一行旁边通过窗口函数算出一个“站在这一行看邻域”的值。典型问题包括每条订单对应的近10单平均价格、每个用户从登录到当前时刻的累计消费、每条日志前面5条日志的状态。这么说吧Group Windows像“期末成绩单”学期结束统一统计出一次成绩Over Windows更像“记账本”每花一笔钱余额立刻重新计算。两种思路各有归宿但Over窗口在实时场景下有个优势不需要等窗口结束数据到达即可算输出延迟更低。3.2 ROWS与RANGE两种边界方式的一次说清Over窗口的边界用ROWS和RANGE定义。ROWS按物理行数算比如“前10行到当前行”不关心时间跨度RANGE按时间范围算比如“当前时间往前15分钟”要求排序字段必须和时间相关。SQL标准里的写法大家都见过-- 最近10条包含当前行在内共11条 SUM(amount) OVER ( PARTITION BY user_id ORDER BY ts ROWS BETWEEN 10 PRECEDING AND CURRENT ROW ) -- 最近15分钟 SUM(amount) OVER ( PARTITION BY user_id ORDER BY ts RANGE BETWEEN INTERVAL 15 MINUTE PRECEDING AND CURRENT ROW )这里有个常见数错个数的点ROWS BETWEEN 10 PRECEDING AND CURRENT ROW包含了当前行和之前10行一共11行。如果只想看“不包括当前行、纯前10行”要写ROWS BETWEEN 10 PRECEDING AND 1 PRECEDING。Flink对FOLLOWING的支持比较受限当前行之后通常只支持到CURRENT ROW如果你需要窗口一直延伸到当前行之后大概率要自己用连接去实现。3.3 SQL和Table API两种写法SQL写法如上Table API也可以写核心是over()和.as()。旧版本和新版本的API差别不小我以常见的Flink 1.13~1.17中间版本为例Over w Over .partitionBy(user_id) .orderBy(ts) .preceding(lit(10).rows()) .as(sum_amt); Table result table .window(w) .select(user_id, ts, amount, sum_amt);RANGE写法把preceding(lit(10).rows())换成preceding(lit(15).minutes())即可。注意Table API里Over的别名as(sum_amt)是给窗口命名的不是给聚合结果命名的所以preceding(10.rows())和.as(sum_amt)一起用select里引用的是窗口别名。新手常在这里写错报错信息又不太直观。3.4 Over窗口的三个硬性要求第一必须ORDER BY一个时间字段。Flink要求Over窗口的排序字段必须是时间属性TIMESTAMP(3)类型的Event Time或Processing Time字段否则会报不支持。这是因为Over窗口在流上需要按序处理没有时间顺序就没法界定“之前”和“之后”。第二RANGE边界的时间间隔必须跟排序字段类型匹配。排序字段是TIMESTAMP(3)间隔就要用INTERVAL排序字段是数字比如秒时间戳部分版本的RANGE会要求用数字边界。这个类型不匹配的报错往往很晚才出现写DDL时就把时间字段统一成TIMESTAMP(3)是最省事的。第三Over窗口的PARTITION BY不是必须的但几乎总是建议写。不写等价于在全表上做Over每个全局分区自成一路数据倾斜和复杂度都会放大。大部分业务都是“按用户、按商品、按设备”分区的所以别省这一行。4. 时间字段是窗口的地基Event Time和Processing Time声明4.1 建表DDL里声明Rowtime和Watermark不管用哪种窗口前提都是表里有一个时间字段。直接读Kafka或文件时原始数据里通常只有业务时间戳需要显式声明成Event Time并定义水位线。Flink DDL里常见写法CREATE TABLE user_actions ( user_id STRING, action_time STRING, amount DOUBLE, ts AS TO_TIMESTAMP(action_time), WATERMARK FOR ts AS ts - INTERVAL 5 SECOND, PRIMARY KEY (user_id) NOT ENFORCED ) WITH ( connector kafka, topic user_actions, properties.bootstrap.servers localhost:9092, format json, properties.group.id flink-window-note );这里最关键的是ts AS TO_TIMESTAMP(action_time)把字符串转成TIMESTAMP(3)然后WATERMARK FOR ts AS ts - INTERVAL 5 SECOND声明允许5秒乱序。如果你没有WATERMARKEvent Time窗口基本不会按预期触发因为Flink不知道什么时候“认为”一个窗口的数据齐了。4.2 用Proctime做Processing Time窗口有些场景不需要业务时间只要系统当前时间比如“每分钟统计目前收到的数据量”可以直接声明Proctime字段CREATE TABLE live_actions ( user_id STRING, action STRING, proc_time AS PROCTIME() ) WITH ( connector kafka, topic live_actions, format json );proc_time AS PROCTIME()是生成列不需要来源字段。之后不管是TUMBLE(proc_time, INTERVAL 1 MINUTE)还是OVER (ORDER BY proc_time ...)都可以直接用。Processing Time窗口简单、无乱序问题但结果不代表真实业务时间只能用于“当下这一刻”的统计。4.3 时间字段选错了会怎样我在项目里见过几次看起来逻辑没错、但结果就是不对的情况最后都查到了时间字段上。把STRING类型时间字段放进TUMBLEFlink会报“Window time field must be of type TIMESTAMP”之类错误这种还算友好。用了Event Time但没声明Watermark窗口永远不触发数据全积压日志里没有明显报错这是最隐蔽的坑。建表时声明了ts AS TO_TIMESTAMP(...)但原始时间戳是13位毫秒数需要TO_TIMESTAMP_LTZ(ts / 1000, 3)不然把毫秒当秒转时间直接错乱。所以我的习惯是在SELECT里先输出一下时间字段肉眼确认时间对不对再做窗口聚合。这一步能省下很多排查工时。5. 模拟项目把几种窗口组合到一张表上做用户行为分析5.1 场景设计与模拟数据为了把前面几章串起来我用一个叫“模拟项目X”的例子假设业务方要给运营看实时大屏需要三块指标每分钟页面访问量、每5分钟看过去1小时活跃用户数、每条访问记录旁边附上该用户最近10条平均浏览时长和最近15分钟累计消费金额。输入是一张用户行为表user_actions字段如下字段类型说明user_idSTRING用户IDaction_timeSTRING发生时间如2024-06-01 10:30:00actionSTRING浏览/下单/支付durationBIGINT本条访问时长秒amountDOUBLE订单金额浏览时为0DDL里把action_time转成ts TIMESTAMP(3)Watermark延迟5秒。为了测试我先往Kafka里塞100条模拟数据10个用户时间从10:00到10:59浏览为主部分下单支付。这种数据量不大但足够验证三种窗口的语义。5.2 窗口聚合分钟级PV和5分钟窗口活跃用户分钟级PV直接用Tumble窗口SELECT TUMBLE_START(ts, INTERVAL 1 MINUTE) AS win_start, COUNT(*) AS pv FROM user_actions GROUP BY TUMBLE(ts, INTERVAL 1 MINUTE);每5分钟看过去1小时活跃用户数用Hop窗口SELECT HOP_START(ts, INTERVAL 5 MINUTE, INTERVAL 1 HOUR) AS win_start, HOP_END(ts, INTERVAL 5 MINUTE, INTERVAL 1 HOUR) AS win_end, COUNT(DISTINCT user_id) AS active_users FROM user_actions GROUP BY HOP(ts, INTERVAL 5 MINUTE, INTERVAL 1 HOUR);这个查询在流模式下会有状态更新滑动窗口每5分钟输出一次但同一个窗口还没真正结束时可能因为迟到数据发生更新。如果Sink是打印到控制台能看到I和-U/U两种消息I是新增-U/U是更新说明Flink认为这个结果是可更新的。选Sink前必须确认它支持更新模式这是模拟项目里最容易被忽略的。5.3 Over聚合最近10条平均浏览时长和15分钟累计金额Over窗口适合逐行计算。这里我想给每条访问记录附上两个值该用户最近10条访问的平均时长以及最近15分钟累计金额。两个查询可以写成一条SQL用子查询或直接叠加Over窗口SELECT user_id, action_time, duration, amount, AVG(duration) OVER ( PARTITION BY user_id ORDER BY ts ROWS BETWEEN 10 PRECEDING AND CURRENT ROW ) AS avg_dur_10, SUM(amount) OVER ( PARTITION BY user_id ORDER BY ts RANGE BETWEEN INTERVAL 15 MINUTE PRECEDING AND CURRENT ROW ) AS sum_amt_15m FROM user_actions;这条SQL跑起来后每条输入都会实时算出一个结果输出行数等于输入行数。注意两个Over窗口共用了同样的PARTITION BY user_id ORDER BY ts但边界不同一个ROWS一个RANGE。Flink允许同一条SQL里出现多个Over窗口但排序字段要一致才高效否则可能增加额外排序。这里的sum_amt_15m用的是RANGE只统计当前时间之前15分钟内的数据忽略时间范围外的行这跟“最近10条”是完全不同的口径。5.4 结果落到外部存储把查询结果写到外部存储时要考虑更新模式。Group Windows的结果如果不带分组键通常可以当作append-only如果带分组键或用了Hop窗口结果会带更新适合用支持Upsert的Sink比如MySQL用primary key的方式。Over窗口的结果也是逐行追加的用Kafka或打印基本都能接。一个稳妥的做法是给结果表里增加一个唯一键。例如Tumble窗口结果可以用TUMBLE_START(ts, INTERVAL 1 MINUTE)作为主键Hop窗口结果用win_start和win_end组合键这样Sink端能正确处理更新。我见过纯粹因为没定义主键导致MySQL结果表出现重复行的案例排查起来很费劲。6. 常见报错、排查实录与避坑技巧6.1 高频报错速查表下面这些报错和问题是我在实际使用中反复碰到、或确认过同行的真实案例整理成速查表现象常见原因解决方案窗口函数报“Window time field must be TIMESTAMP”时间字段是STRING或BIGINT没转换DDL里通过生成列转TO_TIMESTAMP(...)或TO_TIMESTAMP_LTZ(...)Event Time窗口不触发结果一直为空没声明Watermark或Watermark计算有误检查DDL的WATERMARK FOR ts AS ...并观察Watermark是否推进GROUP BY报“Column not found”窗口字段没有同时出现在GROUP BY里确认GROUP BY TUMBLE(ts, ...)中包含整个窗口表达式Over窗口报“ORDER BY must be time attribute”ORDER BY普通字段而非时间字段把排序字段换成Event Time/Processing Time字段RANGE窗口报类型不匹配排序字段是TIMESTAMP但边界用了数字或反之统一用INTERVAL N MINUTE配TIMESTAMP(3)结果出现重复行Sink不支持更新模式且没设主键结果表加唯一键或用Upsert Sink滑动窗口结果数量明显偏大把滑动步长和窗口长度写反了检查HOP(time, slide, size)slide在前size在后6.2 窗口不触发、结果少数据的排查思路这类问题排查起来最费劲我的固定排查顺序是这样的。第一先看时间字段。把ts直接放进SELECT打印出来确认是时间格式、年份月份正确。如果看到时间整体偏移几小时多半是时区或毫秒转秒的问题。第二看Watermark有没有推进。SQL客户端里可以用SHOW WATERMARKS或者直接观察输出延迟。如果Watermark一直停在初始值检查上游数据源的时间字段是否真的在新数据里变大有没有出现长时间没有数据的分区空闲分区会导致Watermark不推进。第三看窗口时间粒度。Flink窗口默认基于UTC毫秒计算DDL里时间已经转成TIMESTAMP(3)但如果客户端配置的table.local-time-zone不对窗口边界可能和你本地时间对不上。窗口结果看似“晚了一小时”或“早了一小时”先检查时区。第四看流的更新语义。有些查询结果正常但写进Sink后数据对不上原因在Sink不支持更新。把Sink换成print或console看日志通常能一眼发现-U和U消息这说明结果在更新。6.3 我自己常用的几条避坑习惯最后说几个长期实践沉淀下来的操作习惯不算什么高深东西但确实帮我少踩很多坑。第一小数据优先。写任何窗口SQL前先用一条静态或限量数据在SQL Client里跑一遍确认窗口边界和辅助函数的值符合预期。不要一上来就接Kafka全量流排错成本高到让你怀疑人生。第二把时间字段统一成一个约定。全项目都要求TIMESTAMP(3)原始字符串是yyyy-MM-dd HH:mm:ss就用TO_TIMESTAMP是13位毫秒就用TO_TIMESTAMP_LTZ(ts / 1000, 3)避免每个任务各写各的时间转换。第三优先用SQL而不是Table API DSL。Table API的Over窗口链式写法在不同版本之间变化大SQL表达窗口语义更标准后面维护的人看到SQL至少不用翻版本文档。第四窗口别嵌套使用。Flink不支持在一个窗口结果上直接再开另一个Group Window如果必须做“窗口之上再开窗口”一般先落中间结果到消息队列再起一个任务消费这是主流做法。我自己在实际使用中的体会是Group Windows和Over Windows在Table API里只是语法上的分类本质都依赖时间字段。窗口边界、时间类型、更新模式这三个点弄清楚了大部分问题都能自己定位。最后分享一个小技巧调试窗口SQL时把窗口辅助函数如TUMBLE_START、TUMBLE_ROWTIME一起SELECT出来既能确认窗口边界又能直接看到每条结果对应的时间范围这个输出比任何日志都好用。