摘要： 在Flink1.13版本中，提出了窗口表值函数（Window TVF）的实现，用于替代旧版的窗口分组（group window）语法，极大简化了Flink SQL代码量，同时提高了执行性能。1 案例需求分析 1.1 案例需求
利用FlinkSQL统计分析每小时求卖得最⽕的3件商品。
1.2 业务分析1. 每小时计算一次指标，所以得用滚动窗口（窗口长度1小时）
2. 最火的3件商品，显然就是求商品交易次数的TOP N，所以需要根据商品分组，因此是一个分组滚动窗口，每一个窗口包含一个商品ID每小时的订单数据，求交易次数，即count(*)。输出(商品ID，window_time，交易次数)
3. 每个窗口输出了一个商品ID当前1小时的成交次数，接下来按照window_time分组、组内按照交易次数排序， 取交易次数最多的N个即可
1.3 准备数据源测试数据文件路径为/home/hadoop/test/orders/orders，样例数据如下所示：
叶修洁,100,3,2022-02-17 16:18:20
卢智宸,200,1,2022-02-17 16:18:22
廖天翊,300,2,2022-02-17 16:18:24
谢煜城,200,5,2022-02-17 16:18:29
程乐驹,200,1,2022-02-17 16:18:33
龚擎宇,200,10,2022-02-17 16:18:37
石煜城,300,1,2022-02-17 16:18:42
金楷瑞,300,1,2022-02-17 16:18:46
田烨磊,500,2,2022-02-17 16:18:50
杜浩宇,400,7,2022-02-17 16:18:54
1.4 建表SET 'execution.target' = 'local';
SET 'sql-client.execution.result-mode' = 'tableau';
CREATE TABLE orders (
`user` STRING,
productId BIGINT,
amount INT,
orderTp TIMESTAMP(0),
WATERMARK FOR orderTp AS orderTp - INTERVAL '1' SECOND
) WITH (
'connector' = 'filesystem',
'path' = '/home/hadoop/test/orders/orders',
'format' = 'csv'
);
desc orders;
SELECt * FROM orders;
2 基于Group windows实现这里直接用Flink SQL来实现，Table API的写法类似就不重复了，为了演示效果我们就每10秒计算一次Top 2的热门商品：
1. 每小时计算一次指标，所以得用滚动窗口（窗口长度1小时）
2. 最火的3件商品，显然就是求商品交易次数的TOP N，所以需要根据商品分组，因此是一个分组滚动窗口，每个窗口包含一个商品ID每小时的订单数据，求交易次数，即count(*)。输出(商品ID，window_time，交易次数)
3. 每个窗口输出了一个商品ID当前1小时的成交次数，接下来按照window_time分组、组内按照交易次数排序， 取交易次数最多的N个即可
2.1 SQL分步骤实现2.1.1 创建source表
CREATE TABLE orders (
`user` STRING,
productId BIGINT,
amount INT,
orderTp TIMESTAMP(0),
WATERMARK FOR orderTp AS orderTp - INTERVAL '1' SECOND
)
2.1.2 统计时间滚动窗口商品交易次数
利用Flink SQL实现滚动时间窗口，窗口中计算商品交易次数。
CREATE VIEW pcntwindow AS
select
productId,
count(productId) as pcnt,
TUMBLE_END(orderTp, INTERVAL '10' SECOND) AS window_end
from orders
GROUP BY
TUMBLE(orderTp, INTERVAL '10' SECOND),
productId;
2.1.3 统计时间滚动窗口每个商品交易次数排名
利用FLink SQL按照窗口结束时间分组，并按交易次数排序后排名(注意不是拿窗口结束时间当做时间属性的处理哈，就是基于它分个组是没问题的)
CREATE VIEW pcntrank AS
select
productId,
pcnt,
window_end,
ROW_NUMBER() over (partition by window_end order by pcnt DESC) as rownum
from pcntwindow;
2.1.4 统计交易次数最多的TOPN商品
利用Flink SQL查询语句直接统计交易次数最多的Top N商品。
select * from pcntrank WHERe rownum <= 2;
2.2 一个SQL搞定的实现select *
from (
select
productId,
pcnt,
window_end,
ROW_NUMBER() over (partition by window_end order by pcnt DESC) as rownum
from (
select
productId,
count(productId) as pcnt,
TUMBLE_END(orderTp, INTERVAL '10' SECOND) AS window_end
from orders
GROUP BY
TUMBLE(orderTp, INTERVAL '10' SECOND),
productId
) )
WHERe rownum <= 2;
Flink SQL内置函数解释如下：
3 基于Window TVF实现Window TVF实现Window Top N语法更简洁，性能更好。
3.1 Window TVF实现语法SELECt [column_list]
FROM (
SELECt [column_list],
ROW_NUMBER() OVER (PARTITION BY window_start, window_end [, col_key1...]
ORDER BY col1 [asc|desc][, col2 [asc|desc]...]) AS rownum
FROM table_name) -- table name替换为window TVF即可
WHERe rownum <= N [AND conditions]
3.2 Window TVF实现案例需求select *
from (
select productId,pcnt,window_end,ROW_NUMBER() over (partition by window_start,
window_end order by pcnt DESC) as rownum
from(
SELECt productId,window_start,window_end,count(productId) as pcnt
FROM TABLE(TUMBLE(TABLE orders, DESCRIPTOR(orderTp), INTERVAL '10' SECOND))
GROUP by productId,window_start, window_end
) )
WHERe rownum <= 2;
注意：Window TopN 要求 PARTITION BY 子句包含Window TVF 的开始和结束列；将来，如果Window TVF 是TUMBLE 或 HOP，我们还可以简化 PARTITION BY 子句以仅包含开始或结束列。
