【大数据】Flink SQL 语法篇（三）：窗口聚合（TUMBLE、HOP、SESSION、CUMULATE）

2024-01-31 12:12
文章标签 sql 数据 语法 窗口 session flink 聚合 database hop tumble cumulate

本文主要是介绍【大数据】Flink SQL 语法篇（三）：窗口聚合（TUMBLE、HOP、SESSION、CUMULATE），希望对大家解决编程问题提供一定的参考价值，需要的开发者们随着小编来一起学习吧！

Flink SQL 语法篇（三）：窗口聚合

  • 1.滚动窗口（TUMBLE）
    • 1.1 Group Window Aggregation 方案（支持 Batch / Streaming 任务）
    • 1.2 Windowing TVF 方案（1.13 只支持 Streaming 任务）
  • 2.滑动窗口（HOP）
    • 2.1 Group Window Aggregation 方案（支持 Batch / Streaming 任务）
    • 2.2 Windowing TVF 方案（1.13 只支持 Streaming 任务）
  • 3.会话窗口（SESSION）
    • 3.1 Group Window Aggregation 方案（支持 Batch / Streaming 任务）
  • 4.渐进式窗口（CUMULATE）
    • 4.1 Windowing TVF 方案（1.13 只支持 Streaming 任务）
  • 5.Window TVF 支持 Grouping Sets、Rollup、Cube

在这里插入图片描述

1.滚动窗口（TUMBLE）

滚动窗口 将每个元素指定给 指定窗口大小 的窗口。滚动窗口具有固定大小，且不重叠。例如，指定一个大小为 5 分钟的滚动窗口。在这种情况下，Flink 将每隔 5 分钟开启一个新的窗口，其中每一条数都会划分到唯一的一个 5 分钟的窗口中，如下图所示。

在这里插入图片描述

  • ⭐ 应用场景：常见的按照一分钟对数据进行聚合。例如，计算一分钟内 PV，UV 数据。
  • ⭐ 实际案例：简单且常见的分维度分钟级别同时在线用户数、总销售额。

那么上面这个案例的 SQL 要咋写呢？

关于滚动窗口，在 1.13 版本之前和 1.13 及之后版本有两种 Flink SQL 实现方式，分别是：

  • Group Window Aggregation（1.13 之前只有此类方案，此方案在 1.13 及之后版本已经标记为废弃，不推荐小伙伴萌使用）。
  • Windowing TVF（1.13 及之后建议使用 Windowing TVF）。

1.1 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
select dim,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)) * 1000  as window_start
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)，第一个参数为 事件时间的时间戳；第二个参数为 滚动窗口大小。

在这里插入图片描述

1.2 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,pv BIGINT,sum_price BIGINT,max_price BIGINT,min_price BIGINT,uv BIGINT,window_start bigint
) WITH ('connector' = 'print'
);-- 数据处理逻辑
insert into sink_table
SELECT dim,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 中。

这个实时任务也是 24 小时一直在运行的，所有的算子在同一时刻都是处于 running 状态的。

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

2.滑动窗口（HOP）

滑动窗口 也是将元素指定给固定长度的窗口。与滚动窗口功能一样，也有窗口大小的概念。不一样的地方在于，滑动窗口有另一个参数控制窗口计算的频率（滑动窗口滑动的步长）。因此，如果滑动的步长小于窗口大小，则滑动窗口之间每个窗口是可以重叠。在这种情况下，一条数据就会分配到多个窗口当中。举例，有 10 分钟大小的窗口，滑动步长为 5 分钟。这样，每 5 分钟会划分一次窗口，这个窗口包含的数据是过去 10 分钟内的数据，如下图所示。
在这里插入图片描述

  • ⭐ 应用场景：比如计算同时在线的数据，要求结果的输出频率是 1 分钟一次，每次计算的数据是过去 5 分钟的数据（有的场景下用户可能在线，但是可能会 2 分钟不活跃，但是这也要算在同时在线数据中，所以取最近 5 分钟的数据就能计算进去了）。
  • ⭐ 实际案例：简单且常见的分维度分钟级别同时在线用户数，1 分钟输出一次，计算最近 5 分钟的数据。

2.1 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)) * 1000 as window_start, 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)。其中：

  • 第一个参数为 事件时间的时间戳。
  • 第二个参数为 滑动窗口的滑动步长。
  • 第三个参数为 滑动窗口大小。

2.2 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
SELECT dim,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 Gap）内没有新的数据出现，则 Session 就会窗口关闭。
在这里插入图片描述

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

目前 1.13 版本中 Flink SQL 不支持 Session 窗口的 Window TVF，所以这里就只介绍 Group Window Aggregation 方案。

3.1 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
SELECT dim,UNIX_TIMESTAMP(CAST(session_start(row_time, interval '5' minute) AS STRING)) * 1000 as window_start, 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 任务）：渐进式窗口在其实就是 固定窗口间隔内提前触发的的滚动窗口，其实就是 Tumble Window + early-fire 的一个事件时间的版本。例如，从每日零点到当前这一分钟绘制累积 UV，其中 10:00 时的 UV 表示从 00:00 到 10:00 的 UV 总数。渐进式窗口可以认为是首先开一个最大窗口大小的滚动窗口，然后根据用户设置的触发的时间间隔将这个滚动窗口拆分为多个窗口，这些窗口具有相同的窗口起点和不同的窗口终点。如下图所示：
在这里插入图片描述

  • ⭐ 应用场景：周期内累计 PV，UV 指标（如每天累计到当前这一分钟的 PV，UV）。这类指标是一段周期内的累计状态，对分析师来说更具统计分析价值，而且几乎所有的复合指标都是基于此类指标的统计（不然离线为啥都要累计一天的数据，而不要一分钟累计的数据呢）。
  • ⭐ 实际案例：每天的截止当前分钟的累计 money（sum(money)），去重 id 数（count(distinct id)）。每天代表渐进式窗口大小为 1 天，分钟代表渐进式窗口移动步长为分钟级别。

明细输入数据：

timeidmoney
2021-11-01 00:01:00A3
2021-11-01 00:01:00B5
2021-11-01 00:01:00A7
2021-11-01 00:02:00C3
2021-11-01 00:03:00C10

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

timecount distinct idsum money
2021-11-01 00:01:00215
2021-11-01 00:02:00318
2021-11-01 00:03:00328

转化为折线图长这样：

在这里插入图片描述
可以看到，其特点就在于，每一分钟的输出结果都是当天零点累计到当前的结果。

渐进式窗口目前只有 Windowing TVF 方案支持。

4.1 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.price.min' = '1','fields.price.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
SELECT UNIX_TIMESTAMP(CAST(window_end AS STRING)) * 1000 as window_end, window_start, sum(money) as sum_money,count(distinct 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 起来，这样写起来非常麻烦，而且会导致一个数据源读取多遍。

这时，有离线 Hive SQL 使用经验的小伙伴萌就会想到，如果有了 Grouping Sets，我们就可以直接用 Grouping Sets 将维度组合写在一条 SQL 中，写起来方便并且执行效率也高。当然，Flink 支持这个功能。

但是目前 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
SELECT UNIX_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 BY window_start, window_end,-- grouping sets 写法GROUPING SETS ((), (age), (sex), (age, sex));

Flink SQL 中 Grouping Sets 的语法和 Hive SQL 的语法有一些不同，如果我们使用 Hive SQL 实现上述 SQL 的语义，其实现如下：

insert into sink_table
SELECT UNIX_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)
);

这篇关于【大数据】Flink SQL 语法篇（三）：窗口聚合（TUMBLE、HOP、SESSION、CUMULATE）的文章就介绍到这儿，希望我们推荐的文章对编程师们有所帮助！


http://www.chinasem.cn/article/663741。

相关文章

详谈redis跟数据库的数据同步问题

详谈redis跟数据库的数据同步问题

《详谈redis跟数据库的数据同步问题》文章讨论了在Redis和数据库数据一致性问题上的解决方案,主要比较了先更新Redis缓存再更新数据库和先更新数据库再更新Redis缓存两种方案,文章指出,删除R... 目录一、Redis 数据库数据一致性的解决方案1.1、更新Redis缓存、删除Redis缓存的区别二
阅读更多...
Redis事务与数据持久化方式

Redis事务与数据持久化方式

《Redis事务与数据持久化方式》该文档主要介绍了Redis事务和持久化机制,事务通过将多个命令打包执行,而持久化则通过快照（RDB）和追加式文件（AOF）两种方式将内存数据保存到磁盘,以防止数据丢失... 目录一、Redis 事务1.1 事务本质1.2 数据库事务与redis事务1.2.1 数据库事务1.
阅读更多...
Mysql 中的多表连接和连接类型详解

Mysql 中的多表连接和连接类型详解

《Mysql中的多表连接和连接类型详解》这篇文章详细介绍了MySQL中的多表连接及其各种类型,包括内连接、左连接、右连接、全外连接、自连接和交叉连接,通过这些连接方式,可以将分散在不同表中的相关数据... 目录什么是多表连接？1. 内连接（INNER JOIN）2. 左连接（LEFT JOIN 或 LEFT
阅读更多...
Oracle Expdp按条件导出指定表数据的方法实例

Oracle Expdp按条件导出指定表数据的方法实例

《OracleExpdp按条件导出指定表数据的方法实例》：本文主要介绍Oracle的expdp数据泵方式导出特定机构和时间范围的数据,并通过parfile文件进行条件限制和配置,文中通过代码介绍... 目录1.场景描述 2.方案分析3.实验验证 3.1 parfile文件3.2 expdp命令导出4.总结
阅读更多...
更改docker默认数据目录的方法步骤

更改docker默认数据目录的方法步骤

《更改docker默认数据目录的方法步骤》本文主要介绍了更改docker默认数据目录的方法步骤,文中通过示例代码介绍的非常详细,对大家的学习或者工作具有一定的参考学习价值,需要的朋友们下面随着小编来一... 目录1.查看docker是否存在并停止该服务2.挂载镜像并安装rsync便于备份3.取消挂载备份和迁
阅读更多...
不删数据还能合并磁盘? 让电脑C盘D盘合并并保留数据的技巧

不删数据还能合并磁盘? 让电脑C盘D盘合并并保留数据的技巧

《不删数据还能合并磁盘?让电脑C盘D盘合并并保留数据的技巧》在Windows操作系统中，合并C盘和D盘是一个相对复杂的任务，尤其是当你不希望删除其中的数据时，幸运的是，有几种方法可以实现这一目标且在... 在电脑生产时，制造商常为C盘分配较小的磁盘空间，以确保软件在运行过程中不会出现磁盘空间不足的问题。但在
阅读更多...
mysql重置root密码的完整步骤(适用于5.7和8.0)

mysql重置root密码的完整步骤(适用于5.7和8.0)

《mysql重置root密码的完整步骤(适用于5.7和8.0)》：本文主要介绍mysql重置root密码的完整步骤,文中描述了如何停止MySQL服务、以管理员身份打开命令行、替换配置文件路径、修改... 目录第一步：先停止mysql服务，一定要停止！方式一：通过命令行关闭mysql服务方式二：通过服务项关闭
阅读更多...
SQL Server数据库磁盘满了的解决办法

SQL Server数据库磁盘满了的解决办法

《SQLServer数据库磁盘满了的解决办法》系统再正常运行,我还在操作中,突然发现接口报错,后续所有接口都报错了,一查日志发现说是数据库磁盘满了,所以本文记录了SQLServer数据库磁盘满了的解... 目录问题解决方法删除数据库日志设置数据库日志大小问题今http://www.chinasem.cn天发
阅读更多...
mysql主从及遇到的问题解决

mysql主从及遇到的问题解决

《mysql主从及遇到的问题解决》本文详细介绍了如何使用Docker配置MySQL主从复制,首先创建了两个文件夹并分别配置了`my.cnf`文件,通过执行脚本启动容器并配置好主从关系,文中还提到了一些... 目录mysql主从及遇到问题解决遇到的问题说明总结mysql主从及遇到问题解决1.基于mysql
阅读更多...
Java如何接收并解析HL7协议数据

Java如何接收并解析HL7协议数据

《Java如何接收并解析HL7协议数据》文章主要介绍了HL7协议及其在医疗行业中的应用,详细描述了如何配置环境、接收和解析数据,以及与前端进行交互的实现方法,文章还分享了使用7Edit工具进行调试的经... 目录一、前言二、正文1、环境配置2、数据接收：HL7Monitor3、数据解析：HL7Busines
阅读更多...

Copyright China编程 23002807@qq.com

沪ICP备2023033072号-2