关于申请网站建设,WordPress主题使用域名加密,google打开wordpress,浙江有限公司网站文章目录
Flink SQL的窗口操作
一、窗口的概述
二、Group Windows
1、滚动窗口#xff08;TUMBLE#xff09;
2、滑动窗口#xff08;HOP#xff09;
3、Session 窗口#xff08;SESSIONTUMBLE
2、滑动窗口HOP
3、Session 窗口SESSION
4、渐进式窗口CUMULATE
三、Over Windows
1、时间区间聚合RANGE OVER Window
2、行数聚合 Flink SQL的窗口操作
一、窗口的概述
在流处理应用中数据是连续不断的因此我们不可能等到所有数据都到了才开始处理。当然我们可以每来一个消息就处理一次但是有时我们需要做一些聚合类的处理例如在过去的1分钟内有多少用户点击了我们的网页。在这种情况下我们必须定义一个窗口用来收集最近一分钟内的数据并对这个窗口内的数据进行计算。
Flink 认为 Batch 是 Streaming 的一个特例所以 Flink 底层引擎是一个流式引擎在上面实现了流处理和批处理。而窗口window就是从 Streaming 到 Batch 的一个桥梁。
一个Window代表有限对象的集合。一个窗口有一个最大的时间戳该时间戳意味着在其代表的某时间点——所有应该进入这个窗口的元素都已经到达。Window就是用来对一个无限的流设置一个有限的集合在有界的数据集上进行操作的一种机制。在Table API和SQL中主要有两种窗口Group Windows 和 Over Windows。 Group Windows 根据时间或行计数间隔将组行聚合成有限的组并对每个组计算一次聚合函数Over Windows 窗口内聚合为每个输入行在其相邻行范围内计算一个聚合 二、Group Windows
1、滚动窗口TUMBLE
滚动窗口定义滚动窗口将每个元素指定给指定窗口大小的窗口。滚动窗口具有固定大小且不重叠。例如指定一个大小为 5 分钟的滚动窗口。在这种情况下Flink 将每隔 5 分钟开启一个新的窗口其中每一条数都会划分到唯一一个 5 分钟的窗口中如下图所示。 TUMBLE函数基于时间属性字段为关系的每一行指定一个窗口。
在流模式下时间属性字段必须是事件或处理时间属性。
在批处理模式下窗口表函数的时间属性字段必须是TIMESTAMP或TIMESTAMP _LTZ类型的属性。TUMBLE的返回值是一个新的关系它包括原始关系的所有列以及另外3列“window_start”、“window_end”、“window_time”以指示指定的窗口。原始时间属性“timecol”将是窗口TVF之后的常规时间戳列。
TUMBLE函数接受三个必需参数一个可选参数
TUMBLE(TABLE data, DESCRIPTOR(timecol), size [, offset ])
data: 是一个表参数可以是与时间属性列的任何关系。timecol: 是一个列描述符指示数据的哪些时间属性列应映射到翻转窗口。size: 是指定滚动窗口宽度的持续时间。offset: 是一个可选参数用于指定窗口起始位置的偏移量。
应用场景常见的按照一分钟对数据进行聚合计算一分钟内 PVUV 数据。
实际案例简单且常见的分维度分钟级别同时在线用户数、总销售额
那么上面这个案例的 SQL 要咋写呢
关于滚动窗口在 1.14 版本之前和 1.14 及之后版本有两种 Flink SQL 实现方式分别是
Group Window Aggregation1.14 之前只有此类方案此方案在 1.14 及之后版本已经标记为废弃不推荐使用Windowing TVF1.14 及之后建议使用 Windowing TVF
这里两种方法都会介绍
Group Window Aggregation 方案支持 Batch\Streaming 任务
使用socket演示
监听9999端口:nc -lk 9999
测试数据如下创建完表启动查询任务后再进行输入
1,12189,80729,2021-05-23 05:16:39
1,78750,7434,2021-05-23 05:16:41
1,38905,75583,2021-05-23 05:16:42
1,29388,52138,2021-05-23 05:16:45
1,51810,84241,2021-05-23 05:16:47
1,34713,87372,2021-05-23 05:16:48
1,62264,61675,2021-05-23 05:16:52
1,32460,29190,2021-05-23 05:17:40
1,73052,15170,2021-05-23 05:23:00 进入阿里云Flink开发平台创建表,首先使用处理时间进行演示(代码如下)
CREATE TABLE tumble_group_proctime ( dim STRING, user_id BIGINT, price BIGINT,timestamp STRING,row_time AS TO_TIMESTAMP(timestamp),pt AS PROCTIME()
) WITH (connector socket,hostname 128.66.209.119, port 9999,format csv
);
查询语句如下,可以看到 Group Window Aggregation 滚动窗口的 SQL 语法就是把 tumble window 的声明写在了 group by 子句中即 tumble(pt, interval 5 second)。第一个参数为处理时间第二个参数为滚动窗口大小。
如果使用窗口开始时间或者结束的话语法如下tumble_start(pt, interval 5 second)、tumble_end(pt, interval 5 second)。
select --窗口开始时间tumble_start(pt, interval 5 second) as window_start,--窗口结束时间tumble_end(pt, interval 5 second) as window_end,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
from tumble_group_proctime
group by
dim,tumble(pt, interval 5 second);
选中查询代码点击调试查看结果。目前没有传入数据所以没有结果。 下面开始通过socket传入测试数据。首先传入第一条数据在几秒后可以看到一条结果。
同时传入剩下的数据可以看到另一条结果 第一个条结果的pv为1即只有一条数据。这里因为在这个5秒的窗口时间内只有第一条数据。第二个结果的pv为8即有八条数据。显然在这个5秒的时间窗口内有八条数据。这条结果是八条数据聚合的结果。
接下来使用事件时间进行演示代码如下 --事件时间演示
CREATE TABLE tumble_group_eventime ( dim STRING, user_id BIGINT, price BIGINT,timestamp STRING,row_time AS TO_TIMESTAMP(timestamp),watermark for row_time as row_time - interval 0 second
) WITH (connector socket,hostname 128.66.209.119, port 9999,format csv
);
查询语句如下
select --窗口开始时间tumble_start(row_time, interval 5 second) as window_start,--窗口结束时间tumble_end(row_time, interval 5 second) as window_end,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
from tumble_group_eventime
group bydim,tumble(row_time, interval 5 second);
选中查询代码点击调试查看结果。目前没有传入数据所以没有结果。 下面开始通过socket传入测试数据。首先传入第一条数据发现无论等待多久都没有结果产生。
这是因为第一条数据输入后创建了一个时间窗口。但是没有输入第二条数据即没有出现数据的事件时间大于窗口的结束时间所以这个窗口不会结束因此就不会触发计算也就看不到结果。
输入第二条数据发现看到了结果 观察测试数据可以看到第一条数据的事件时间为2021-05-23 05:16:39处于上面结果的窗口时间内。第二条数据的事件时间为2021-05-23 05:16:41大于上面结果的窗口的结束时间因而触发了这个窗口的计算。
输入剩下的数据可以看到结果。 可以对比数据分析结果这里不再赘述。
Window TVF 方案1.14 只支持 Streaming 任务1.15版本开始支持 Batch\Streaming 任务
建表语句
-- TVF
CREATE TABLE tumble_tvf ( dim STRING, user_id BIGINT, price BIGINT,timestamp STRING,row_time AS TO_TIMESTAMP(timestamp),watermark for row_time as row_time - interval 0 second
) WITH (connector socket,hostname 179.18.59.99, port 9999,format csv
); 查询语句
SELECT dim,window_start,window_end,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 tumble_tvf, DESCRIPTOR(row_time), INTERVAL 5 SECOND))
GROUP BY window_start, window_end,dim;
可以看到 Windowing TVF 滚动窗口的写法就是把 tumble window 的声明写在了数据源的 Table 子句中即 TABLE(TUMBLE(TABLE source_table, DESCRIPTOR(row_time), INTERVAL 5 SECOND))包含三部分参数。
第一个参数 TABLE source_table 声明数据源表第二个参数 DESCRIPTOR(row_time) 声明数据源的时间戳第三个参数 INTERVAL 5 SECOND 声明滚动窗口大小为 5 秒。
返回值除了输入的参数外还有window_start, window_end,window_time。window_start返回窗口的起始时间包含边界window_end返回窗口的结束时间包含边界,window_time返回窗口的结束时间不包含边界。window_time等于window_end减去1ms。
注意事项 事件时间中滚动窗口的窗口计算触发是由 Watermark 推动的。 2、滑动窗口HOP
滑动窗口定义滑动窗口也是将元素指定给固定长度的窗口。与滚动窗口功能一样也有窗口大小的概念。不一样的地方在于滑动窗口有另一个参数控制窗口计算的频率滑动窗口滑动的步长。因此如果滑动的步长小于窗口大小则滑动窗口之间每个窗口是可以重叠。在这种情况下一条数据就会分配到多个窗口当中。举例有 10 分钟大小的窗口滑动步长为 5 分钟。这样每 5 分钟会划分一次窗口这个窗口包含的数据是过去 10 分钟内的数据如下图所示。 在流模式下时间属性字段必须是事件或处理时间属性。
在批处理模式下窗口表函数的时间属性字段必须是TIMESTAMP或TIMESTAMP _LTZ类型的属性。TUMBLE的返回值是一个新的关系它包括原始关系的所有列以及另外3列“window_start”、“window_end”、“window_time”以指示指定的窗口。原始时间属性“timecol”将是窗口TVF之后的常规时间戳列。
HOP接受四个必需参数一个可选参数
HOP(TABLE data, DESCRIPTOR(timecol), slide, size [, offset ])
data: 是一个表参数可以是与时间属性列的任何关系。timecol是一个列描述符指示数据的哪些时间属性列应映射到跳跃窗口。slide: 是一个持续时间指定顺序跳跃窗口开始之间的持续时间size: 是指定跳跃窗口宽度的持续时间。offset: 是一个可选参数用于指定窗口起始位置的偏移量。
应用场景比如计算同时在线的数据要求结果的输出频率是 1 分钟一次每次计算的数据是过去 5 分钟的数据有的场景下用户可能在线但是可能会 2 分钟不活跃但是这也要算在同时在线数据中所以取最近 5 分钟的数据就能计算进去了
实际案例简单且常见的分维度分钟级别同时在线用户数2 秒钟输出一次计算最近 6 秒钟的数据
依然是 Group Window Aggregation、Windowing TVF 两种方案
Group Window Aggregation 方案支持 Batch\Streaming 任务 CREATE TABLE hop_group ( dim STRING, user_id BIGINT, price BIGINT,timestamp STRING,row_time AS TO_TIMESTAMP(timestamp),watermark for row_time as row_time - interval 0 second
) WITH (connector socket,hostname 178.23.141.244, port 9999,format csv
);SELECT hop_start(row_time, interval 2 SECOND, interval 6 SECOND) as window_start,hop_end(row_time, interval 2 SECOND, interval 6 SECOND) as window_end, dim,count(distinct user_id) as uv
FROM hop_group
GROUP BY dim, hop(row_time, interval 2 SECOND, interval 6 SECOND);
可以看到 Group Window Aggregation 滚动窗口的写法就是把 hop window 的声明写在了 group by 子句中即 hop(row_time, interval 2 SECOND, interval 6 SECOND)。其中
第一个参数为事件时间的时间戳
第二个参数为滑动窗口的滑动步长
第三个参数为滑动窗口大小。
查询结果 Windowing TVF 方案1.14 只支持 Streaming 任务1.15版本开始支持 Batch\Streaming 任务 CREATE TABLE hop_tvf ( dim STRING, user_id BIGINT, price BIGINT,timestamp STRING,row_time AS TO_TIMESTAMP(timestamp),watermark for row_time as row_time - interval 0 second
) WITH (connector socket,hostname 178.23.141.244, port 9999,format csv
);SELECT dim,window_start, window_end, count(distinct user_id) as uv
FROM TABLE(HOP(TABLE hop_tvf, DESCRIPTOR(row_time), interval 2 SECOND, interval 6 SECOND))
GROUP BY window_start, window_end,dim;
可以看到 Windowing TVF 滚动窗口的写法就是把 hop window 的声明写在了数据源的 Table 子句中即 TABLE(HOP(TABLE source_table, DESCRIPTOR(row_time), INTERVAL 2 SECOND, INTERVAL 6 SECOND))包含四部分参数
第一个参数 TABLE source_table 声明数据源表
第二个参数 DESCRIPTOR(row_time) 声明数据源的时间戳
第三个参数 INTERVAL 2 SECOND 声明滚动窗口滑动步长大小为2 SECOND。
第四个参数 INTERVAL 6 SECOND 声明滚动窗口大小为6 SECOND。
查询结果 3、Session 窗口SESSION
Session 窗口定义Session 时间窗口和滚动、滑动窗口不一样其没有固定的持续时间如果在定义的间隔期Session Gap内没有新的数据出现则 Session 就会窗口关闭。如下图对比所示 应用场景计算每个用户在活跃期间一个 Session总共购买的商品数量如果用户 5 分钟没有活动则视为 Session 断开
案例
目前 1.15 版本中 Flink SQL 不支持 Session 窗口的 Window TVF所以这里就只介绍 Group Window Aggregation 方案
Group Window Aggregation 方案支持 Batch\Streaming 任务 CREATE TABLE session_group ( dim STRING, user_id BIGINT, price BIGINT,timestamp STRING,row_time AS TO_TIMESTAMP(timestamp),watermark for row_time as row_time - interval 0 second
) WITH (connector socket,hostname 178.23.148.244, port 9999,format csv
);SELECT session_start(row_time, interval 5 SECOND) as window_start, session_end(row_time, interval 5 SECOND) as window_end, dim,count(1) as pv
FROM session_group
GROUP BY dim, session(row_time, interval 5 SECOND); 注意事项 上述 SQL 任务是在整个 Session 窗口结束之后才会把数据输出。Session 窗口即支持 处理时间 也支持 事件时间。但是处理时间只支持在 Streaming 任务中运行Batch 任务不支持。 Session gap 间隔是5s实际上是不包含5s即大于5s才会触发计算
可以看到 Group Window Aggregation 中 Session 窗口的写法就是把 session window 的声明写在了 group by 子句中即 session(row_time, interval 5 SECOND)。其中
第一个参数为事件时间的时间戳
第二个参数为 Session gap 间隔。 4、渐进式窗口CUMULATE 渐进式窗口定义渐进式窗口是 固定窗口间隔内提前触发的的滚动窗口其实就是 Tumble Window early-fire 的一个事件时间的版本。
例如从每日零点到当前这一分钟绘制累积 UV其中 10:00 时的 UV 表示从 00:00 到 10:00 的 UV 总数。
渐进式窗口可以认为是首先开一个最大窗口大小的滚动窗口然后根据用户设置的触发的时间间隔将这个滚动窗口拆分为多个窗口这些窗口具有相同的窗口起点和不同的窗口终点。如下图所示 这些CUMULATE函数根据时间属性列分配窗口。
在流模式下时间属性字段必须是事件或处理时间属性。
在批处理模式下窗口表函数的时间属性字段必须是TIMESTAMP或TIMESTAMP _LTZ类型的属性。CUMULATE的返回值是一个新的关系它包括原始关系的所有列以及另外3列“window_start”、“window_end”、“window_time”以指示指定的窗口。原始时间属性“timecol”将是窗口TVF之后的常规时间戳列。
CUMULATE接受四个必需参数一个可选参数
CUMULATE(TABLE data, DESCRIPTOR(timecol), step, size)
data: 是一个表参数可以是与时间属性列的任何关系。timecol是一个列描述符指示数据的哪些时间属性列应映射到累积窗口。step: 是指定连续累积窗口结束之间增加的窗口大小的持续时间。size: 是指定累积窗口的最大宽度的持续时间。size必须是step的整数倍。offset: 是一个可选参数用于指定窗口起始位置的偏移量。
应用场景周期内累计 PVUV 指标如每天累计到当前这一分钟的 PVUV。这类指标是一段周期内的累计状态对分析师来说更具统计分析价值而且几乎所有的复合指标都是基于此类指标的统计不然离线为啥都要累计一天的数据而不要一分钟累计的数据呢。
实际案例每天的截止当前分钟的累计 moneysum(money)去重 id 数count(distinct id)。每天代表渐进式窗口大小为 1 天分钟代表渐进式窗口移动步长为分钟级别。举例如下
明细输入数据 time id money 2021-11-01 00:01:00 A 3 2021-11-01 00:01:00 B 5 2021-11-01 00:01:00 A 7 2021-11-01 00:02:00 C 3 2021-11-01 00:03:00 C 10
预期经过渐进式窗口计算的输出数据 time count distinct id sum money 2021-11-01 00:01:00 2 15 2021-11-01 00:02:00 3 18 2021-11-01 00:03:00 3 28
转化为折线图长这样 可以看到其特点就在于每一分钟的输出结果都是当天零点累计到当前的结果。
渐进式窗口目前只有 Windowing TVF 方案支持
Windowing TVF 方案1.14 只支持 Streaming 任务1.15版本开始支持 Batch\Streaming 任务 CREATE TABLE cumulate_tvf (-- 用户 iduser_id BIGINT,-- 金额money BIGINT,-- 事件时间戳row_time AS cast(CURRENT_TIMESTAMP as timestamp(3)),-- watermark 设置WATERMARK FOR row_time AS row_time - INTERVAL 0 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
);SELECT window_start, window_end,sum(money) as sum_money,count(distinct user_id) as count_distinct_id
FROM TABLE(CUMULATE(TABLE cumulate_tvf, 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 天到了第二天新开一个窗口重新累计。 三、Over Windows
OVER Window over聚合是传统数据库的标准开窗不同于Group WindowOVER窗口中每1个元素都对应1个窗口。OVER窗口可以按照实际元素的行或实际的元素值时间戳值确定窗口因此流数据元素可能分布在多个窗口中。
在应用OVER窗口的流式数据中每1个元素都对应1个OVER窗口。每1个元素都触发1次数据计算每个触发计算的元素所确定的行都是该元素所在窗口的最后1行。在实时计算的底层实现中OVER窗口的数据进行全局统一管理数据只存储1份逻辑上为每1个元素维护1个OVER窗口为每1个元素进行窗口计算完成计算后会清除过期的数据。
拿 over聚合 与 窗口聚合 做一个对比其之间的直观不同之处在于
窗口聚合不在 group by 中的字段不能直接在 select 中拿到Over 聚合能够保留原始字段
Over 聚合的语法总结如下
SELECTagg1(col1) OVER (definition1) AS colName,...aggN(colN) OVER (definition1) AS colNameN
FROM Tab1;
其中
agg1(col1)按照GROUP BY指定col1列对输入数据进行聚合计算。OVER (definition1)OVER窗口定义。AS colName别名。
按照计算行的定义方式OVER Window可以分为以下两类
RANGE OVER Window具有相同时间值的所有元素行视为同一计算行即具有相同时间值的所有行都是同一个窗口。
ROWS OVER Window每1行元素都被视为新的计算行即每1行都是一个新的窗口。 1、时间区间聚合RANGE OVER Window
窗口数据
RANGE OVER Window所有具有共同元素值元素时间戳的元素行确定一个窗口。
窗口语法
SELECTagg1(col1) OVER([PARTITION BY (value_expression1,..., value_expressionN)]ORDER BY timeColRANGE BETWEEN (UNBOUNDED | timeInterval) PRECEDING AND CURRENT ROW) AS colName,
...
FROM Tab1;
value_expression进行分区的字表达式。timeCol元素排序的时间字段。timeInterval定义根据当前行开始向前追溯指定时间的元素行。
案例每条数据对应最近 1 小时的区间即最新输出的一条数据的 sum 聚合结果就是最近一小时数据的 amount 之和。
CREATE TABLE over_window_time (order_id BIGINT,product BIGINT,amount BIGINT,order_time as cast(CURRENT_TIMESTAMP as TIMESTAMP(3)),WATERMARK FOR order_time AS order_time - INTERVAL 0.001 SECOND
) WITH (connector datagen,rows-per-second 1,fields.order_id.min 1,fields.order_id.max 2,fields.amount.min 1,fields.amount.max 10,fields.product.min 1,fields.product.max 2
);SELECT product, order_time, amount,SUM(amount) OVER (PARTITION BY productORDER BY order_time-- 标识统计范围是一个 product 的最近 1 小时的数据RANGE BETWEEN INTERVAL 1 HOUR PRECEDING AND CURRENT ROW) AS one_hour_prod_amount_sum
FROM over_window_time;
结果如下 2、行数聚合
窗口数据
ROWS OVER Window的每个元素都确定一个窗口。
窗口语法
SELECTagg1(col1) OVER([PARTITION BY (value_expression1,..., value_expressionN)]ORDER BY timeColROWS BETWEEN (UNBOUNDED | rowCount) PRECEDING AND CURRENT ROW) AS colName, ...FROM Tab1;
value_expression分区值表达式。timeCol元素排序的时间字段。rowCount定义根据当前行开始向前追溯几行元素。
案例最新输出的一条数据的 sum 聚合结果是最近 5 行数据的 amount 之和。
CREATE TABLE over_window_row (order_id BIGINT,product BIGINT,amount BIGINT,order_time as cast(CURRENT_TIMESTAMP as TIMESTAMP(3)),WATERMARK FOR order_time AS order_time - INTERVAL 0.001 SECOND
) WITH (connector datagen,rows-per-second 1,fields.order_id.min 1,fields.order_id.max 2,fields.amount.min 1,fields.amount.max 2,fields.product.min 1,fields.product.max 2
);SELECT product, order_time, amount,SUM(amount) OVER (PARTITION BY productORDER BY order_time-- 标识统计范围是一个 product 的最近 5 行数据ROWS BETWEEN 5 PRECEDING AND CURRENT ROW) AS five_rows_amount_sum
FROM over_window_row;
结果如下 博客主页https://lansonli.blog.csdn.net欢迎点赞 收藏 ⭐留言 如有错误敬请指正本文由 Lansonli 原创首发于 CSDN博客停下休息的时候不要忘了别人还在奔跑希望大家抓紧时间学习全力奔赴更美好的生活✨
