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

本文来自互联网用户投稿，该文观点仅代表作者本人，不代表本站立场。本站仅提供信息存储空间服务，不拥有所有权，不承担相关法律责任。如若转载，请注明出处：http://www.hqwc.cn/news/162007.html

如若内容造成侵权/违法违规/事实不符，请联系编程知识网进行投诉反馈email:809451989@qq.com，一经查实，立即删除！

相关文章

django REST框架- Django-ninja

django REST框架- Django-ninja

Django 是我学习的最早的web框架，大概在2014年，当时选他原因也很简单就是网上资料比较丰富，自然是遇到问题更容易找答案，直到 2018年真正开始拿django做项目，才对他有了更全面的了解。他是一个入门有门槛，学…
阅读更多...
【触想智能】工业显示器上市前的检测项目分享

【触想智能】工业显示器上市前的检测项目分享

工业显示器在上市前，需要做一项重要的工作，那就是工业显示器出厂前的产品可靠性检测。 工业显示器选择的测试项目相比商用端更为严格，常见的性能测试项目包括高温老化、防尘防水、电磁静电干扰、防摔防撞等，在工业级应用领域&…
阅读更多...
使用自定义函数拟合辨识HPPC工况下的电池数据（适用于一阶RC、二阶RC等电池模型）

使用自定义函数拟合辨识HPPC工况下的电池数据（适用于一阶RC、二阶RC等电池模型）

该程序可以离线辨识HPPC工况下的电池数据，只需要批量导入不同SOC所对应的脉冲电流电压数据，就可以瞬间获得SOC为[100% 90% 80% 70% 60% 50% 40% 30% 20% 10% 0%]的所有电池参数,迅速得到参数辨识的结果并具有更高的精度，可以很大程度上降低参…
阅读更多...
CentOS7查看和关闭防火墙

CentOS7查看和关闭防火墙

文章目录 CentOS7查看和关闭防火墙1.CentOS 7 默认使用的是firewall作为防火墙2.查看防火墙状态3.停止firewall4.禁止firewall开机启动5.开启防火墙开机启动6.开启端口7.移除端口8.重启防火墙9.查看某个端口是否开启10.查询开放列表总结:systemctl常用命令总结:firewalld-cmd配…
阅读更多...
keil代码编辑区配色方案

keil代码编辑区配色方案

第一步找到global.prop文件打开 ### 第二步复制下面的文本替换global.prop的内容，保存。 # properties for all file types indent.automatic1 virtual.space0 view.whitespace0 view.endofline0 code.page936 caretline.visible1 highlight.matchingbraces1 prin…
阅读更多...
如何用 GPT-4 全模式（All Tools）帮你高效学习和工作？

如何用 GPT-4 全模式（All Tools）帮你高效学习和工作？

「十项全能」的 ChatGPT ，用起来感受如何？ 之前，作为 ChatGPT Plus 用户，如果你集齐下面这五个模式，就会成为别人羡慕的对象。 但现在，人们更加期盼的，是下面这个提示的出现： 这个提…
阅读更多...
c++ 信息学奥赛 2047：【例5.16】过滤空格

c++ 信息学奥赛 2047：【例5.16】过滤空格

#include<cstdio> using namespace std; char st[200]; int main() { while (scanf("%s",&st)1)printf("%s ",st); //%s 后要有一个空格&#xff0c;不能省略return 0; } 解析&#xff1a;本题中使用一个技巧&#xff0c;那就是scanf函数…
阅读更多...
使用 Visio 绘制立方体

使用 Visio 绘制立方体

一、自带的立方体 点击&#xff0c;将其拖拽至空白画布中 点击立方体&#xff0c;可以拖拽立方体的边线改变形状&#xff0c;如下&#xff1a; 可以看到&#xff0c;立方体三个面的颜色是不同的&#xff0c;最上方的面颜色浅一些。 如果想要三个面的颜色相同&#xff0c;先点击…
阅读更多...
Python基础入门例程45-NP45 禁止重复注册（条件语句）

Python基础入门例程45-NP45 禁止重复注册（条件语句）

最近的博文&#xff1a; Python基础入门例程44-NP44 判断列表是否为空&#xff08;条件语句&#xff09;-CSDN博客 Python基础入门例程43-NP43 判断布尔值&#xff08;条件语句&#xff09;-CSDN博客 Python基础入门例程42-NP42 公式计算器&#xff08;运算符&#xff09;-C…
阅读更多...
【0基础】全网最细致【PYTHON入门】教学

【0基础】全网最细致【PYTHON入门】教学

前言 可以说很多人学编程&#xff0c;不玩点爬虫确实少了很多意思&#xff0c;不管是业余、接私活还是职业爬虫&#xff0c;爬虫世界确实挺精彩的。 今天来给大家浅谈一下爬虫&#xff0c;目的是让准备学爬虫或者刚开始起步的小伙伴们&#xff0c;对爬虫有一个更深更全的认知…
阅读更多...
NLP学习笔记：使用 Python 进行NLTK

NLP学习笔记：使用 Python 进行NLTK

一、说明 本文和接下来的几篇文章将介绍 Python NLTK 库。NLTK — 自然语言工具包 — NLTK 是一个强大的开源库&#xff0c;用于 NLP 的研究和开发。它内置了 50 多个文本语料库和词汇资源。它支持文本标记化、词性标记、词干提取、词形还原、命名实体提取、分割、分类、语义推…
阅读更多...
计算机网络第4章-IPv4

计算机网络第4章-IPv4

IPv4数据报格式 IPv4数据报格式如下图所示 其中&#xff0c;有如下的关键字段需要特别注意&#xff1a; 版本&#xff08;号&#xff09;&#xff1a; 版本字段共4比特&#xff0c;规定了数据报的IP协议版本。通过查看版本号吗&#xff0c;路由器能确定如何解释IP数据报的剩…
阅读更多...
推荐文章
最新文章

Copyright @ 2022~2023