Flink SQL 窗口聚合详解

1.滚动窗⼝（TUMBLE）

**滚动窗⼝定义：**滚动窗⼝将每个元素指定给指定窗⼝⼤⼩的窗⼝，滚动窗⼝具有固定⼤⼩，且不重叠。

例如，指定⼀个⼤⼩为 5 分钟的滚动窗⼝，Flink 将每隔 5 分钟开启⼀个新的窗⼝，其中每⼀条数都会划分到唯⼀⼀个 5 分钟的窗⼝中。

在这里插入图片描述

**应⽤场景：**按照⼀分钟对数据进⾏聚合，计算⼀分钟内 PV，UV 数据。

**实际案例：**分维度分钟级别统计在线⽤户数、总销售额。

滚动窗⼝在 1.13 版本之前和 1.13 及版本之后有两种 Flink SQL 实现⽅式

Group Window Aggregation（1.13 之前）和 Windowing TVF（1.13 及之后）

Group Window Aggregation ⽅案（⽀持 Batch\Streaming 任务）：

-- 数据源表
CREATE TABLE source_table (-- 维度数据dim STRING,-- ⽤户 iduser_id BIGINT,-- ⽤户price BIGINT,-- 事件时间戳row_time AS cast(CURRENT_TIMESTAMP as timestamp(3)),-- watermark 设置WATERMARK FOR row_time AS row_time - INTERVAL '5' SECOND
) WITH ('connector' = 'datagen','rows-per-second' = '10','fields.dim.length' = '1','fields.user_id.min' = '1','fields.user_id.max' = '100000','fields.price.min' = '1','fields.price.max' = '100000'
)-- 数据汇表
CREATE TABLE sink_table (dim STRING,pv BIGINT,sum_price BIGINT,max_price BIGINT,min_price BIGINT,uv BIGINT,window_start bigint
) WITH ('connector' = 'print'
)-- 数据处理逻辑
insert into sink_table
selectdim,count(*) as pv,sum(price) as sum_price,max(price) as max_price,min(price) as min_price,-- 计算 uv 数count(distinct user_id) as uv,UNIX_TIMESTAMP(CAST(tumble_start(row_time, interval '1' minute) AS STRING)) * 10
from source_table
group bydim,tumble(row_time, interval '1' minute)

Group Window Aggregation 滚动窗⼝的 SQL 语法，把 tumble window 的声明写在了 group by ⼦句中，即 tumble(row_time, interval ‘1’ minute) ，第⼀个参数为事件时间的时间戳，第⼆个参数为滚动窗⼝⼤⼩。

Window TVF ⽅案（1.13 只⽀持 Streaming 任务）：

-- 数据源表
CREATE TABLE source_table (-- 维度数据dim STRING,-- ⽤户 iduser_id BIGINT,-- ⽤户price BIGINT,-- 事件时间戳row_time AS cast(CURRENT_TIMESTAMP as timestamp(3)),-- watermark 设置WATERMARK FOR row_time AS row_time - INTERVAL '5' SECOND
) WITH ('connector' = 'datagen','rows-per-second' = '10','fields.dim.length' = '1','fields.user_id.min' = '1','fields.user_id.max' = '100000','fields.price.min' = '1','fields.price.max' = '100000'
)-- 数据汇表
CREATE TABLE sink_table (dim STRING,pv BIGINT,sum_price BIGINT,max_price BIGINT,min_price BIGINT,uv BIGINT,window_start bigint
) WITH ('connector' = 'print'
)-- 数据处理逻辑
insert into sink_table
SELECTdim,UNIX_TIMESTAMP(CAST(window_start AS STRING)) * 1000 as window_start,count(*) as pv,sum(price) as sum_price,max(price) as max_price,min(price) as min_price,count(distinct user_id) as uv
FROM TABLE(TUMBLE(TABLE source_table, DESCRIPTOR(row_time), INTERVAL '60' SECOND))
GROUP BY window_start, window_end,dim

Windowing TVF 滚动窗⼝的写法把 tumble window 的声明写在了数据源的 Table ⼦句中，包含三部分参数：

TABLE(
TUMBLE(TABLE source_table, DESCRIPTOR(row_time), INTERVAL '60' SECOND)
)

第⼀个参数 TABLE source_table 声明数据源表；

第⼆个参数 DESCRIPTOR(row_time) 声明数据源的时间戳字段；

第三个参数 INTERVAL ‘60’ SECOND 声明滚动窗⼝⼤⼩为 1 min。

实时场景 SQL 语义：假设 Orders 为 kafka，target_table 也为 Kafka，这个 SQL ⽣成的实时任务，在执⾏时，会⽣成三个算⼦。

数据源算⼦（From Order）：

连接到 Kafka topic，数据源算⼦⼀直运⾏，实时的从 Order Kafka 中⼀条⼀条的读取数据，然后⼀条⼀条发送给下游的窗⼝聚合算⼦

窗⼝聚合算⼦（TUMBLE 算⼦）：

接收到上游算⼦发的⼀条⼀条的数据，然后将每⼀条数据按照时间戳划分到对应的窗⼝中（根据事件时间、处理时间的不同语义进⾏划分），上述案例为事件时间，事件时间中，滚动窗⼝算⼦接收到上游的 Watermark ⼤于窗⼝的结束时间时，则说明当前这⼀分钟的滚动窗⼝已经结束了，将窗⼝计算完的结果发往下游算⼦（⼀条⼀条发给下游数据汇算⼦）

数据汇算⼦（INSERT INTO target_table）：

接收到上游发的⼀条⼀条的数据，写⼊到 target_table Kafka 中

注意： 事件时间中滚动窗⼝的窗⼝计算触发是由 Watermark 推动的。

2.滑动窗⼝（HOP）

**滑动窗⼝定义：**滑动窗⼝是将元素指定给固定⻓度的窗⼝，与滚动窗⼝功能⼀样，也有窗⼝⼤⼩的概念，不⼀样的地⽅在于，滑动窗⼝有另⼀个参数控制窗⼝计算的频率（滑动窗⼝滑动的步⻓），如果滑动的步⻓⼩于窗⼝⼤⼩，则滑动窗⼝之间每个窗⼝是可以重叠，在这种情况下，⼀条数据就会分配到多个窗⼝当中。

**举例：**有 10 分钟⼤⼩的窗⼝，滑动步⻓为 5 分钟，每 5 分钟会划分⼀次窗⼝，这个窗⼝包含的数据是过去 10 分钟内的数据。

在这里插入图片描述

**应⽤场景：**计算同时在线的数据，要求结果的输出频率是 1 分钟⼀次，每次计算的数据是过去 5 分钟的数据（有的场景下⽤户可能在线，但是可能会 2 分钟不活跃，但是这也要算在同时在线数据中，所以取最近 5 分钟的数据就能计算进去了）

**实际案例：**分维度分钟级别同时在线⽤户数，1 分钟输出⼀次，计算最近 5 分钟的数据，Group Window Aggregation、Windowing TVF 两种⽅案

Group Window Aggregation ⽅案（⽀持 Batch\Streaming 任务）：

CREATE TABLE source_table (-- 维度数据dim STRING,-- ⽤户 iduser_id BIGINT,-- ⽤户price BIGINT,-- 事件时间戳row_time AS cast(CURRENT_TIMESTAMP as timestamp(3)),-- watermark 设置WATERMARK FOR row_time AS row_time - INTERVAL '5' SECOND
) WITH ('connector' = 'datagen','rows-per-second' = '10','fields.dim.length' = '1','fields.user_id.min' = '1','fields.user_id.max' = '100000','fields.price.min' = '1','fields.price.max' = '100000'
);-- 数据汇表
CREATE TABLE sink_table (dim STRING,uv BIGINT,window_start bigint
) WITH ('connector' = 'print'
);-- 数据处理逻辑
insert into sink_table
SELECT dim,
UNIX_TIMESTAMP(CAST(hop_start(row_time, interval '1' minute, interval '5' minute) AS STRING)) * 10,
count(distinct user_id) as uv
FROM source_table
GROUP BY dim, hop(row_time, interval '1' minute, interval '5' minute)

Group Window Aggregation 滚动窗⼝的写法把 hop window 的声明写在了 group by ⼦句中，即

hop(row_time, interval '1' minute, interval '5' minute)

第⼀个参数为事件时间的时间戳字段；

第⼆个参数为滑动窗⼝的滑动步⻓；

第三个参数为滑动窗⼝⼤⼩。

Windowing TVF ⽅案（1.13 只⽀持 Streaming 任务）：

-- 数据源表
CREATE TABLE source_table (-- 维度数据dim STRING,-- ⽤户 iduser_id BIGINT,-- ⽤户price BIGINT,-- 事件时间戳row_time AS cast(CURRENT_TIMESTAMP as timestamp(3)),-- watermark 设置WATERMARK FOR row_time AS row_time - INTERVAL '5' SECOND
) WITH ('connector' = 'datagen','rows-per-second' = '10','fields.dim.length' = '1','fields.user_id.min' = '1','fields.user_id.max' = '100000','fields.price.min' = '1','fields.price.max' = '100000'
);-- 数据汇表
CREATE TABLE sink_table (dim STRING,uv BIGINT,window_start bigint
) WITH ('connector' = 'print'
);-- 数据处理逻辑
insert into sink_table
SELECTdim,UNIX_TIMESTAMP(CAST(window_start AS STRING)) * 1000 as window_start, count(distinct user_id) as bucket_uv
FROM TABLE(HOP(TABLE source_table, DESCRIPTOR(row_time), INTERVAL '1' MINUTES, INTERVAL '5' MINUTES))
GROUP BY window_start, window_end,dim

Windowing TVF 滑动窗⼝的写法把 hop window 的声明写在了数据源的 Table ⼦句中，即

TABLE(HOP(TABLE source_table, DESCRIPTOR(row_time), INTERVAL '1' MINUTES, INTERVAL '5' MINUTES))

第⼀个参数 TABLE source_table 声明数据源表；

第⼆个参数 DESCRIPTOR(row_time) 声明数据源的时间戳；

第三个参数 INTERVAL ‘1’ MINUTES 声明滚动窗⼝滑动步⻓⼤⼩为 1 min。

第四个参数 INTERVAL ‘5’ MINUTES 声明滚动窗⼝⼤⼩为 5 min。

3.Session 窗⼝

**Session 窗⼝定义：**Session 时间窗⼝和滚动、滑动窗⼝不⼀样，其没有固定的持续时间，如果在定义的间隔期（Session Gap）内没有新的数据出现，则 Session 就会窗⼝关闭。

在这里插入图片描述

**实际案例：**计算每个⽤户在活跃期间（⼀个 Session）总共购买的商品数量，如果⽤户 5 分钟没有活动，则视为 Session 断开

⽬前 1.13 版本中 Flink SQL 不⽀持 Session 窗⼝的 Window TVF，只介绍 Group Window Aggregation ⽅案。

Group Window Aggregation ⽅案（⽀持 Batch\Streaming 任务）：

-- 数据源表，⽤户购买⾏为记录表
CREATE TABLE source_table (-- 维度数据dim STRING,-- ⽤户 iduser_id BIGINT,-- ⽤户price BIGINT,-- 事件时间戳row_time AS cast(CURRENT_TIMESTAMP as timestamp(3)),-- watermark 设置WATERMARK FOR row_time AS row_time - INTERVAL '5' SECOND
) WITH ('connector' = 'datagen','rows-per-second' = '10','fields.dim.length' = '1','fields.user_id.min' = '1','fields.user_id.max' = '100000','fields.price.min' = '1','fields.price.max' = '100000'
);-- 数据汇表
CREATE TABLE sink_table (dim STRING,pv BIGINT, -- 购买商品数量window_start bigint
) WITH ('connector' = 'print'
);-- 数据处理逻辑
insert into sink_table
SELECTdim,UNIX_TIMESTAMP(CAST(session_start(row_time, interval '5' minute) AS STRING)) * 10,count(1) as pv
FROM source_table
GROUP BY dim, session(row_time, interval '5' minute)

**注意：**上述 SQL 任务是在整个 Session 窗⼝结束之后才会把数据输出，Session 窗⼝⽀持处理时间和事件时间，但是处理时间只⽀持在 Streaming 任务中运⾏，Batch 任务不⽀持。

Group Window Aggregation 中 Session 窗⼝的写法把 session window 的声明写在了 group by ⼦句中

session(row_time, interval '5' minute)

第⼀个参数为事件时间的时间戳；

第⼆个参数为 Session gap 间隔。

4.渐进式窗⼝（CUMULATE）

**渐进式窗⼝定义（1.13 只⽀持 Streaming 任务）：**渐进式窗⼝可以认为是⾸先开⼀个最⼤窗⼝⼤⼩的滚动窗⼝，然后根据⽤户设置的触发的时间间隔将这个滚动窗⼝拆分为多个窗⼝，这些窗⼝具有相同的窗⼝起点和不同的窗⼝终点。

**示例：**从每⽇零点到当前这⼀分钟绘制累积 UV，其中 10:00 时的 UV 表示从 00:00 到 10:00 的 UV 总数。

在这里插入图片描述

**应⽤场景：**周期内累计 PV，UV 指标（如每天累计到当前这⼀分钟的 PV，UV），这类指标是⼀段周期内的累计状态。

**实际案例：**每天的截⽌当前分钟的累计 money（sum(money)），去重 id 数（count(distinct id)），每天代表渐进式窗⼝⼤⼩为 1 天，分钟代表渐进式窗⼝移动步⻓为分钟级别。

明细输⼊数据：

在这里插入图片描述

预期经过渐进式窗⼝计算的输出数据：

在这里插入图片描述

**特点：**每⼀分钟的输出结果都是当天零点累计到当前的结果，渐进式窗⼝只有 Windowing TVF ⽅案⽀持。

Windowing TVF ⽅案（1.13 只⽀持 Streaming 任务）

-- 数据源表
CREATE TABLE source_table (-- ⽤户 iduser_id BIGINT,-- ⽤户money BIGINT,-- 事件时间戳row_time AS cast(CURRENT_TIMESTAMP as timestamp(3)),-- watermark 设置WATERMARK FOR row_time AS row_time - INTERVAL '5' SECOND
) WITH ('connector' = 'datagen','rows-per-second' = '10','fields.user_id.min' = '1','fields.user_id.max' = '100000','fields.money.min' = '1','fields.money.max' = '100000'
);-- 数据汇表
CREATE TABLE sink_table (window_end bigint,window_start bigint,sum_money BIGINT,count_distinct_id bigint
) WITH ('connector' = 'print'
);-- 数据处理逻辑
insert into sink_table
SELECTUNIX_TIMESTAMP(CAST(window_end AS STRING)) * 1000 as window_end, window_start, sum(money) as sum_money,count(distinct user_id) as count_distinct_id
FROM TABLE(CUMULATE(TABLE source_table, DESCRIPTOR(row_time), INTERVAL '60' SECOND, INTERVAL '1' DAY))
GROUP BYwindow_start, window_end

Windowing TVF 滚动窗⼝的写法把 cumulate window 的声明写在了数据源的 Table ⼦句中

TABLE(CUMULATE(TABLE source_table,DESCRIPTOR(row_time),INTERVAL '60' SECOND, INTERVAL '1' DAY)
)

第⼀个参数 TABLE source_table 声明数据源表；

第⼆个参数 DESCRIPTOR(row_time) 声明数据源的时间戳；

第三个参数 INTERVAL ‘60’ SECOND 声明渐进式窗⼝触发的渐进步⻓为 1 min。

第四个参数 INTERVAL ‘1’ DAY 声明整个渐进式窗⼝的⼤⼩为 1 天，到了第⼆天新开⼀个窗⼝重新累计。

5.Window TVF ⽀持 Grouping Sets、Rollup、Cube

**应⽤场景：**多个维度组合（cube）计算，把每个维度写⼀遍 union all 起来麻烦⽽且会导致⼀个数据源读取多遍。

⽤ Grouping Sets 将维度组合写在⼀条 SQL 中，⽅便且执⾏效率⾼，⽬前 Grouping Sets 只在 Window TVF 中⽀持，不⽀持 Group Window Aggregation。

**示例：**计算每⽇零点累计到当前这⼀分钟的，分汇总、age、sex、age+sex 维度的⽤户数。

-- ⽤户访问明细表
CREATE TABLE source_table (age STRING,sex STRING,user_id BIGINT,row_time AS cast(CURRENT_TIMESTAMP as timestamp(3)),WATERMARK FOR row_time AS row_time - INTERVAL '5' SECOND
) WITH ('connector' = 'datagen','rows-per-second' = '1','fields.age.length' = '1','fields.sex.length' = '1','fields.user_id.min' = '1','fields.user_id.max' = '100000'
);CREATE TABLE sink_table (age STRING,sex STRING,uv BIGINT,window_end bigint
) WITH ('connector' = 'print'
);insert into sink_table
SELECTUNIX_TIMESTAMP(CAST(window_end AS STRING)) * 1000 as window_end,if (age is null, 'ALL', age) as age,if (sex is null, 'ALL', sex) as sex,count(distinct user_id) as bucket_uv
FROM TABLE(CUMULATE(TABLE source_table, DESCRIPTOR(row_time), INTERVAL '5' SECOND, INTERVAL '1' DAY))
GROUP BYwindow_start, window_end,-- grouping sets 写法GROUPING SETS ((), (age), (sex), (age, sex))

Flink SQL 中 Grouping Sets 的语法和 Hive SQL 的语法有不同，使⽤ Hive SQL 实现上述 SQL 的语义，实现如下：

insert into sink_table
SELECTUNIX_TIMESTAMP(CAST(window_end AS STRING)) * 1000 as window_end, if (age is null, 'ALL', age) as age,if (sex is null, 'ALL', sex) as sex,count(distinct user_id) as bucket_uv
FROM source_table
GROUP BYage, sex
-- hive sql grouping sets 写法
GROUPING SETS ((), (age), (sex), (age, sex)
)