AI 助理
文档备案控制台
开发者社区 数据库 文章 正文

FlinkSQL窗口新特性（Window TVF）

2022-07-10 2599
版权
版权声明：
本文内容由阿里云实名注册用户自发贡献，版权归原作者所有，阿里云开发者社区不拥有其著作权，亦不承担相应法律责任。具体规则请查看《 阿里云开发者社区用户服务协议》和 《阿里云开发者社区知识产权保护指引》。如果您发现本社区中有涉嫌抄袭的内容，填写 侵权投诉表单进行举报，一经查实，本社区将立刻删除涉嫌侵权内容。
简介： 理解Window TVFWindow TVF使用

理解Window TVF


表值函数(table-valued function, TVF)，顾名思义就是指返回值是⼀张表的函数，是不是很神奇，在Oracle、SQL Server等数据库中屡⻅不鲜。⽽在Flink的上⼀个稳定版本1.13中，社区通过FLIP-145提出了窗⼝表值函数(window TVF)的实现，⽤于替代旧版的窗⼝分组(group window)语法。



注意：

  • Window TVF⼀个重要的特性是窗⼝定义函数返回的⼀张表，且包含了window_start, window_end,window_time三个字段分别代表窗⼝开始时间戳、结束时间戳、窗⼝时间戳；

  • window_time是窗⼝ 往下发送数据的时间属性，等于window_end - 1毫秒。

  • 特别注意：window_start 和 window_end 列是常规时间戳列，⽽不是时间属性。 因此它们不能⽤作后续基于时间 的操作中的时间属性。

  • 为了往下游操作传播时间属性，您需要在 GROUP BY ⼦句中额外添加 window_time 列。 window_time 是 Windowing TVF ⽣成的，它是窗⼝的时间属性。 然后后⾯的查询可以使⽤该列进⾏后续的基于时 间的操作（不需要再设置时间属性列；额），例如级联窗⼝聚合和窗⼝TopN。



我们可以通过如下链接来理解⼀下：

https://cwiki.apache.org/confluence/display/FLINK/FLIP-145%253A+Support+SQL+windowing+table-valued+function#FLIP145:SupportSQLwindowingtablevaluedfunction-TumblingWindows



实例



准备⼀个数据⽂件~/test/bid/bids⽂件，⾥⾯存储的是供应商报价信息



2020-04-1508:05,4.00,C,supplier12020-04-1508:07,2.00,A,supplier12020-04-1508:09,5.00,D,supplier22020-04-1508:11,3.00,B,supplier22020-04-1508:13,1.00,E,supplier12020-04-1508:17,6.00,F,supplier2


CREATETABLEbid (
bidtimeTIMESTAMP(3),
priceDECIMAL(10, 2),
itemSTRING,
supplier_idSTRING,
WATERMARKFORbidtimeASbidtime-INTERVAL'1'SECOND) WITH (
'connector'='filesystem',
'path'='/home/hadoop/test/bid',
'format'='csv');
descbid;
SELECT*FROMbid;

滚动窗口

image.png




SELECTwindow_start, window_end, SUM(price) aspriceFROMTABLE(
TUMBLE(TABLEbid, DESCRIPTOR(bidtime), INTERVAL'10'MINUTES))
GROUPBYwindow_start, window_end;
--也可以根据其他业务字段groupbySELECTwindow_start, window_end, supplier_id,SUM(price) aspriceFROMTABLE(
TUMBLE(TABLEbid, DESCRIPTOR(bidtime), INTERVAL'10'MINUTES))
GROUPBYsupplier_id,window_start, window_end;
--还可以获取window_timeSELECTwindow_start, window_end,window_time, SUM(price) aspriceFROMTABLE(
TUMBLE(TABLEbid, DESCRIPTOR(bidtime), INTERVAL'10'MINUTES))
GROUPBYwindow_start, window_end,window_time;


>SELECT*FROMBid;
--------------------------|bidtime|price|item|--------------------------|8:07|$2|A||8:11|$3|B||8:05|$4|C||8:09|$5|D||8:13|$1|E||8:17|$6|F|-------------------------->SELECT*FROMTABLE(
TUMBLE(TABLEBid, DESCRIPTOR(bidtime), INTERVAL'10'MINUTES));
--orwiththenamedparams--note: theDATAparammustbethefirst>SELECT*FROMTABLE(
TUMBLE(
DATA=>TABLEBid,
TIMECOL=>DESCRIPTOR(bidtime),
SIZE=>INTERVAL'10'MINUTES));
------------------------------------------------------|bidtime|price|item|window_start|window_end|------------------------------------------------------|8:07|$2|A|8:00|8:10||8:11|$3|B|8:10|8:20||8:05|$4|C|8:00|8:10||8:09|$5|D|8:00|8:10||8:13|$1|E|8:10|8:20||8:17|$6|F|8:10|8:20|------------------------------------------------------>SELECTwindow_start, window_end, SUM(price)
FROMTABLE(
TUMBLE(TABLEBid, DESCRIPTOR(bidtime), INTERVAL'10'MINUTES))
GROUPBYwindow_start, window_end;
-------------------------------------|window_start|window_end|price|-------------------------------------|8:00|8:10|$11||8:10|8:20|$10|-------------------------------------


滑动窗口

image.png



SELECTwindow_start, window_end, SUM(price) aspriceFROMTABLE(
HOP(TABLEbid, DESCRIPTOR(bidtime), INTERVAL'5'MINUTES, INTERVAL'10'MINUTES))
GROUPBYwindow_start, window_end;
--也可以根据其他业务字段groupbySELECTwindow_start, window_end, supplier_id, SUM(price) aspriceFROMTABLE(
HOP(TABLEbid, DESCRIPTOR(bidtime), INTERVAL'5'MINUTES, INTERVAL'10'MINUTES))
GROUPBYsupplier_id,window_start, window_end;
--还可以获取window_timeSELECTwindow_start, window_end, window_time,SUM(price) aspriceFROMTABLE(
HOP(TABLEbid, DESCRIPTOR(bidtime), INTERVAL'5'MINUTES, INTERVAL'10'MINUTES))
GROUPBYwindow_start, window_end,window_time;




>SELECT*FROMTABLE(
HOP(TABLEBid, DESCRIPTOR(bidtime), INTERVAL'5'MINUTES, INTERVAL'10'MINUTES));
--orwiththenamedparams--note: theDATAparammustbethefirst>SELECT*FROMTABLE(
HOP(
DATA=>TABLEBid,
TIMECOL=>DESCRIPTOR(bidtime),
SLIDE=>INTERVAL'5'MINUTES,
SIZE=>INTERVAL'10'MINUTES));
------------------------------------------------------|bidtime|price|item|window_start|window_end|------------------------------------------------------|8:07|$2|A|8:00|8:10||8:07|$2|A|8:05|8:15||8:11|$3|B|8:05|8:15||8:11|$3|B|8:10|8:20||8:05|$4|C|8:00|8:10||8:05|$4|C|8:05|8:15||8:09|$5|D|8:00|8:10||8:09|$5|D|8:05|8:15||8:13|$1|E|8:05|8:15||8:13|$1|E|8:10|8:20||8:17|$6|F|8:10|8:20||8:17|$6|F|8:15|8:25|------------------------------------------------------>SELECTwindow_start, window_end, SUM(price)
FROMTABLE(
HOP(TABLEBid, DESCRIPTOR(bidtime), INTERVAL'5'MINUTES, INTERVAL'10'MINUTES))
GROUPBYwindow_start, window_end;
--------------------------------------|window_start|window_end|price|--------------------------------------|8:00|8:10|$11||8:05|8:15|$15||8:10|8:20|$10||8:15|8:25|$6|--------------------------------------



累计窗口

Window TVF新增了窗⼝：CUMULATE WINDOW，即累积窗⼝，⽤于计算⼀些累积值，例如：当⽇累积UV。


image.png



SELECTwindow_start, window_end, SUM(price) aspriceFROMTABLE(
CUMULATE(TABLEbid, DESCRIPTOR(bidtime), INTERVAL'2'MINUTES, INTERVAL'10'MINUTES))
GROUPBYwindow_start, window_end;
--也可以根据其他业务字段groupbySELECTwindow_start, window_end, supplier_id,SUM(price) aspriceFROMTABLE(
CUMULATE(TABLEbid, DESCRIPTOR(bidtime), INTERVAL'2'MINUTES, INTERVAL'10'MINUTES))
GROUPBYsupplier_id,window_start, window_end;
--还可以获取window_timeSELECTwindow_start, window_end, window_time,SUM(price) aspriceFROMTABLE(
CUMULATE(TABLEbid, DESCRIPTOR(bidtime), INTERVAL'2'MINUTES, INTERVAL'10'MINUTES))
GROUPBYwindow_start, window_end,window_time;


>SELECT*FROMTABLE(
CUMULATE(TABLEBid, DESCRIPTOR(bidtime), INTERVAL'2'MINUTES, INTERVAL'10'MINUTES));
--orwiththenamedparams--note: theDATAparammustbethefirst>SELECT*FROMTABLE(
CUMULATE(
DATA=>TABLEBid,
TIMECOL=>DESCRIPTOR(bidtime),
STEP=>INTERVAL'2'MINUTES,
SIZE=>INTERVAL'10'MINUTES));
------------------------------------------------------|bidtime|price|item|window_start|window_end|------------------------------------------------------|8:07|$2|A|8:00|8:08||8:07|$2|A|8:00|8:10||8:11|$3|B|8:10|8:12||8:11|$3|B|8:10|8:14||8:11|$3|B|8:10|8:16||8:11|$3|B|8:10|8:18||8:11|$3|B|8:10|8:20||8:05|$4|C|8:00|8:06||8:05|$4|C|8:00|8:08||8:05|$4|C|8:00|8:10||8:09|$5|D|8:00|8:10||8:13|$1|E|8:10|8:14||8:13|$1|E|8:10|8:16||8:13|$1|E|8:10|8:18||8:13|$1|E|8:10|8:20||8:17|$6|F|8:10|8:18||8:17|$6|F|8:10|8:20|------------------------------------------------------>SELECTwindow_start, window_end, SUM(price)
FROMTABLE(
CUMULATE(TABLEBid, DESCRIPTOR(bidtime), INTERVAL'2'MINUTES, INTERVAL'10'MINUTES))
GROUPBYwindow_start, window_end;
--------------------------------------|window_start|window_end|price|--------------------------------------|8:00|8:06|$4||8:00|8:08|$6||8:00|8:10|$11||8:10|8:12|$3||8:10|8:14|$4||8:10|8:16|$4||8:10|8:18|$10||8:10|8:20|$10|--------------------------------------


会话窗口


image.png



>SELECT*FROMBid;
------------------------------------|bidtime|price|item|bidder|------------------------------------|8:07|$2|A|takidau||8:05|$1|A|klk||8:09|$10|B|takidau||8:08|$3|A|klk||8:17|$20|B|klk|------------------------------------>SELECT*FROMTABLE(
SESSION(TABLEBidPARTITIONBYbidder, DESCRIPTOR(bidtime), DESCRIPTOR(bidder), INTERVAL'5'MINUTES);
--orwiththenamedparams--note: theDATAparammustbethefirst>SELECT*FROMSESSION(
data=>TABLEBidsPARTITIONBYbidder,
timecol=>DESCRIPTOR(bidtime),
gap=>INTERVAL'5'MINUTES);
----------------------------------------------------------------|bidtime|price|item|bidder|window_start|window_end|----------------------------------------------------------------|8:07|$2|A|takidau|8:07|8:14||8:05|$1|A|klk|8:05|8:13||8:09|$10|B|takidau|8:07|8:14||8:08|$3|A|klk|8:05|8:13||8:17|$20|B|klk|8:17|8:22|---------------------------------------------------------------->SELECTbidder, window_start, window_end, SUM(price)
FROMTABLE(
SESSION(TABLEBidPARTITIONBYbidder, DESCRIPTOR(bidtime), DESCRIPTOR(bidder), INTERVAL'5'MINUTES)
GROUPBYbidder, window_start, window_end;
-----------------------------------------------|bidder|window_start|window_end|price|-----------------------------------------------|takidau|8:07|8:14|$12||klk|8:05|8:13|$4||klk|8:17|8:22|$20|-----------------------------------------------


窗口应用场景总计

窗口

应用场景

滚动窗口

BI分析(每⼩时的GMV、成交数量、商品成交量排⾏，凡是求每⼀⼩时/每天的某个指标的语意)

滑动窗口

监控、监测分析（最近1⼩时的cpu使⽤率的最⼤值、最⼩值。。。，凡是最近1⼩时/1分钟这种 语意的均可）

会话窗口

⽤户⾏为分析(分析每个⽤户在每个会话期间的点击次数、收藏次数)

累计窗口

固定的开始结束时间，每隔多⻓时间计算⼀次指标(每⼗分钟计算当天的累积UV)


Window TVF多维分析




前⾯的例⼦⾥⾯都是按照所有字段分组，加⼊我们想根据多个分组集合来分组应该怎么办呢？需要写多个SQL单独 执⾏吗？


Window TVF语法的标准化带来了更多的灵活性和扩展性，⽤户可以直接在 window 窗⼝函数上进⾏多维分析。可 以直接进⾏ GROUPING SETS、ROLLUP、CUBE 的分析计算。如果是在 1.13 之前的版本，我们可能需要对这些分 组进⾏单独的 SQL 聚合，再对聚合结果做 union 操作才能达到类似的效果。⽽现在，类似这种多维分析的场景， 可以直接在 window TVF 上支持


需求：


现在想每⼗分钟分别按照supplier_id、item分组求价格平均值，同时要考虑不按照supplier_id、item分组求价格 平均值，还要考虑同时按照supplier_id、item分组求价格平均值：



传统做法：多sql


SELECTwindow_start, window_end, supplier_id, SUM(price) aspriceFROMTABLE(
TUMBLE(TABLEbid, DESCRIPTOR(bidtime), INTERVAL'10'MINUTES))
GROUPBYwindow_start, window_end, supplier_id;
SELECTwindow_start, window_end, item, SUM(price) aspriceFROMTABLE(
TUMBLE(TABLEbid, DESCRIPTOR(bidtime), INTERVAL'10'MINUTES))
GROUPBYwindow_start, window_end, item;
SELECTwindow_start, window_end, SUM(price) aspriceFROMTABLE(
TUMBLE(TABLEbid, DESCRIPTOR(bidtime), INTERVAL'10'MINUTES))
GROUPBYwindow_start, window_end;
SELECTwindow_start, window_end, supplier_id,item,SUM(price) aspriceFROMTABLE(
TUMBLE(TABLEbid, DESCRIPTOR(bidtime), INTERVAL'10'MINUTES))
GROUPBYwindow_start, window_end,supplier_id,item;


使用 window TVF 实现


SELECTwindow_start, window_end, window_time,supplier_id,item, SUM(price) aspriceFROMTABLE(
TUMBLE(TABLEbid, DESCRIPTOR(bidtime), INTERVAL'10'MINUTES))
GROUPBYwindow_start, window_end, window_time,GROUPINGSETS ((supplier_id,item),
(supplier_id),(item),());

注意：window_start, window_end, window_time必须出现在group by 子句中。

文章标签：
实时计算 Flink版
BI
流计算
SQL
关系型数据库
监控
Oracle
数据库
存储
相关实践学习
基于Hologres+Flink搭建GitHub实时数据大屏
通过使用Flink、Hologres构建实时数仓，并通过Hologres对接BI分析工具（以DataV为例），实现海量数据实时分析.
实时计算 Flink 实战课程
如何使用实时计算 Flink 搞定数据处理难题？实时计算 Flink 极客训练营产品、技术专家齐上阵，从开源 Flink功能介绍到实时计算 Flink 优势详解，现场实操，5天即可上手！ 欢迎开通实时计算 Flink 版： https://cn.aliyun.com/product/bigdata/sc Flink Forward Asia 介绍： Flink Forward 是由 Apache 官方授权，Apache Flink Community China 支持的会议，通过参会不仅可以了解到 Flink 社区的最新动态和发展计划，还可以了解到国内外一线大厂围绕 Flink 生态的生产实践经验，是 Flink 开发者和使用者不可错过的盛会。 去年经过品牌升级后的 Flink Forward Asia 吸引了超过2000人线下参与，一举成为国内最大的 Apache 顶级项目会议。结合2020年的特殊情况，Flink Forward Asia 2020 将在12月26日以线上峰会的形式与大家见面。
游客tehculasawvxq
目录
相关文章
灵杰开发者
|
存储 分布式计算 Apache
构建 Streaming Lakehouse：使用 Paimon 和 Hudi 的性能对比
Apache Paimon 和 Apache Hudi 作为数据湖存储格式，有着高吞吐的写入和低延迟的查询性能，是构建数据湖的常用组件。本文将在阿里云EMR 上，针对数据实时入湖场景，对 Paimon 和 Hudi 的性能进行比对，然后分别以 Paimon 和 Hudi 作为统一存储搭建准实时数仓。
灵杰开发者
61107 9
构建 Streaming Lakehouse：使用 Paimon 和 Hudi 的性能对比
阿里云实时计算Flink
|
SQL 双11 流计算
Flink SQL 功能解密系列 —— 流计算“撤回(Retraction)”案例分析
通俗讲retract就是传统数据里面的更新操作，也就是说retract是流式计算场景下对数据更新的处理方式。
阿里云实时计算Flink
20706 10
ha_lydms
|
消息中间件 程序员 API
Flink中时间和窗口
Flink中时间和窗口
ha_lydms
711 0
阿甘兄
|
Java
Mac下安装JDK11（国内镜像）
Mac下安装JDK11（国内镜像）
阿甘兄
9432 0
灵杰开发者
|
11月前
|
人工智能 运维 Java
Flink Agents：基于Apache Flink的事件驱动AI智能体框架
本文基于Apache Flink PMC成员宋辛童在Community Over Code Asia 2025的演讲，深入解析Flink Agents项目的技术背景、架构设计与应用场景。该项目聚焦事件驱动型AI智能体，结合Flink的实时处理能力，推动AI在工业场景中的工程化落地，涵盖智能运维、直播分析等典型应用，展现其在AI发展第四层次——智能体AI中的重要意义。
灵杰开发者
3491 27
Flink Agents：基于Apache Flink的事件驱动AI智能体框架
技术小达人
|
SQL 分布式计算 数据处理
FlinkSQL开发经验分享
FlinkSQL开发经验分享
技术小达人
735 8
申某某
|
缓存 监控 数据处理
Flink 四大基石之窗口（Window）使用详解
在流处理场景中，窗口（Window）用于将无限数据流切分成有限大小的“块”，以便进行计算。Flink 提供了多种窗口类型，如时间窗口（滚动、滑动、会话）和计数窗口，通过窗口大小、滑动步长和偏移量等属性控制数据切分。窗口函数包括增量聚合函数、全窗口函数和ProcessWindowFunction，支持灵活的数据处理。应用案例展示了如何使用窗口进行实时流量统计和电商销售分析。
申某某
2666 28
芯在这
|
SQL 关系型数据库 MySQL
实时计算 Flink版操作报错合集之报错：org.apache.flink.table.api.validationexception如何解决
在使用实时计算Flink版过程中，可能会遇到各种错误，了解这些错误的原因及解决方法对于高效排错至关重要。针对具体问题，查看Flink的日志是关键，它们通常会提供更详细的错误信息和堆栈跟踪，有助于定位问题。此外，Flink社区文档和官方论坛也是寻求帮助的好去处。以下是一些常见的操作报错及其可能的原因与解决策略。
芯在这
1602 1
游客wkxim4agoo6le
|
SQL 存储 应用服务中间件
Flink SQL之ProcessTime与EventTime的使用
笔记
游客wkxim4agoo6le
1754 0
Flink SQL之ProcessTime与EventTime的使用

热门文章

最新文章

  • 1
    阿里云DTS踩坑经验分享系列｜如何使用DTS进行MySQL->ClickHouse同步
  • 2
    nodeJS 的 npm 设置国内高速镜像之淘宝镜像的方法
  • 3
    如何在SpringBoot中集成JWT(JSON Web Token)鉴权
  • 4
    AI加持的阿里云飞天大数据平台技术揭秘
  • 5
    Stable Diffusion原理详解
  • 6
    DFSMN在阿里巴巴的应用以及如何采用开源代码训练DFSMN模型
  • 7
    vi命令小结
  • 8
    ASP.NET CheckBoxList Operations with jQuery
  • 9
    HDOJ(HDU) 2106 decimal system(进制相互转换问题)
  • 10
    Java 集合
  • 1
    阿里云 ESSD AutoPL 云盘完整指南：性能分层、计费规则与控制台操作
    82
  • 2
    同城O2O系统开发趋势分析：本地生活数字化下的新机会
    63
  • 3
    【AI】Agent 全栈进阶｜系统化学习路线专题
    611
  • 4
    创业团队开发聊天应用，需要解决哪些 IM 技术难题？
    90
  • 5
    互联网医院系统开发：HIS接口、电子病历与处方数据同步设计
    82
  • 6
    RAG系统测试实战：如何验证企业知识库AI助手是否可靠？
    97
  • 7
    风控系统如何从“管交易”延伸到“管IP”？IP风险画像落地指南
    98
  • 8
    LLM测试体系设计：从Prompt测试到模型质量评估的完整实践
    90
  • 9
    【AI时代软件项目管理系列】2. 从工具到数字员工：AI 在项目团队中的角色演进
    115
  • 10
    阿里云服务器弹性临时盘是什么？哪些ECS支持弹性临时盘？
    110

    • 相关电子书

    更多
  • 低代码开发师（初级）实战教程
  • 冬季实战营第三期：MySQL数据库进阶实战
  • 阿里巴巴DevOps 最佳实践手册
    • 下一篇
    一条命令迁移，帮你实现 OpenClaw 与 Hermes Agent 记忆互通！