理解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;

滚动窗口

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|-------------------------------------

滑动窗口

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。

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|--------------------------------------

会话窗口

>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 子句中。

FlinkSQL窗口新特性（Window TVF）