大连网站快速建设推荐,开发公司物业服务合同,常见的网站推广途径,0453牡丹江信息网二手车Flink系列之#xff1a;窗口聚合 一、窗口表值函数#xff08;TVF#xff09;聚合二、窗口表值函数TVF三、分组集四、ROLLUP五、CUBE六、选择组窗口开始和结束时间戳七、多级窗口聚合八、分组窗口聚合九、时间属性十、选取分组窗口开始和结束时间戳 一、窗口表值函数#x… Flink系列之窗口聚合 一、窗口表值函数TVF聚合二、窗口表值函数TVF三、分组集四、ROLLUP五、CUBE六、选择组窗口开始和结束时间戳七、多级窗口聚合八、分组窗口聚合九、时间属性十、选取分组窗口开始和结束时间戳 一、窗口表值函数TVF聚合
适用于流批窗口聚合在 GROUP BY 子句中定义包含应用窗口 TVF 的关系的“window_start”和“window_end”列。就像使用常规 GROUP BY 子句的查询一样使用按窗口聚合进行分组的查询将计算每个组的单个结果行。
SELECT ...
FROM windowed_table -- relation applied windowing TVF
GROUP BY window_start, window_end, ...与连续表上的其他聚合不同窗口聚合不会发出中间结果而只会发出最终结果即窗口末尾的总聚合。此外窗口聚合会在不再需要时清除所有中间状态。
二、窗口表值函数TVF
Flink 支持 TUMBLE、HOP 和 CUMULATE 类型的窗口聚合。在流模式下窗口表值函数的时间属性字段必须位于事件或处理时间属性上。在批处理模式下窗口表值函数的时间属性字段必须是 TIMESTAMP 或 TIMESTAMP_LTZ 类型的属性。
以下是 TUMBLE、HOP 和 CUMULATE 窗口聚合的一些示例。
-- 表必须具有时间属性例如该表中的“bidtime”
Flink SQL desc Bid;
-----------------------------------------------------------------------------------------
| name | type | null | key | extras | watermark |
-----------------------------------------------------------------------------------------
| bidtime | TIMESTAMP(3) *ROWTIME* | true | | | bidtime - INTERVAL 1 SECOND |
| price | DECIMAL(10, 2) | true | | | |
| item | STRING | true | | | |
| supplier_id | STRING | true | | | |
-----------------------------------------------------------------------------------------Flink SQL SELECT * FROM Bid;
--------------------------------------------
| bidtime | price | item | supplier_id |
--------------------------------------------
| 2020-04-15 08:05 | 4.00 | C | supplier1 |
| 2020-04-15 08:07 | 2.00 | A | supplier1 |
| 2020-04-15 08:09 | 5.00 | D | supplier2 |
| 2020-04-15 08:11 | 3.00 | B | supplier2 |
| 2020-04-15 08:13 | 1.00 | E | supplier1 |
| 2020-04-15 08:17 | 6.00 | F | supplier2 |
---------------------------------------------- 翻转窗口聚合
Flink SQL SELECT window_start, window_end, SUM(price)FROM TABLE(TUMBLE(TABLE Bid, DESCRIPTOR(bidtime), INTERVAL 10 MINUTES))GROUP BY window_start, window_end;
-------------------------------------------
| window_start | window_end | price |
-------------------------------------------
| 2020-04-15 08:00 | 2020-04-15 08:10 | 11.00 |
| 2020-04-15 08:10 | 2020-04-15 08:20 | 10.00 |
--------------------------------------------- 跳跃窗口聚合
Flink SQL SELECT window_start, window_end, SUM(price)FROM TABLE(HOP(TABLE Bid, DESCRIPTOR(bidtime), INTERVAL 5 MINUTES, INTERVAL 10 MINUTES))GROUP BY window_start, window_end;
-------------------------------------------
| window_start | window_end | price |
-------------------------------------------
| 2020-04-15 08:00 | 2020-04-15 08:10 | 11.00 |
| 2020-04-15 08:05 | 2020-04-15 08:15 | 15.00 |
| 2020-04-15 08:10 | 2020-04-15 08:20 | 10.00 |
| 2020-04-15 08:15 | 2020-04-15 08:25 | 6.00 |
--------------------------------------------- 累积窗口聚合
Flink SQL SELECT window_start, window_end, SUM(price)FROM TABLE(CUMULATE(TABLE Bid, DESCRIPTOR(bidtime), INTERVAL 2 MINUTES, INTERVAL 10 MINUTES))GROUP BY window_start, window_end;
-------------------------------------------
| window_start | window_end | price |
-------------------------------------------
| 2020-04-15 08:00 | 2020-04-15 08:06 | 4.00 |
| 2020-04-15 08:00 | 2020-04-15 08:08 | 6.00 |
| 2020-04-15 08:00 | 2020-04-15 08:10 | 11.00 |
| 2020-04-15 08:10 | 2020-04-15 08:12 | 3.00 |
| 2020-04-15 08:10 | 2020-04-15 08:14 | 4.00 |
| 2020-04-15 08:10 | 2020-04-15 08:16 | 4.00 |
| 2020-04-15 08:10 | 2020-04-15 08:18 | 10.00 |
| 2020-04-15 08:10 | 2020-04-15 08:20 | 10.00 |
-------------------------------------------注意为了更好地理解窗口的行为我们简化了时间戳值的显示不显示尾随零例如如果类型为 TIMESTAMP(3)2020-04-15 08:05 在 Flink SQL Client 中应显示为 2020-04-15 08:05:00.000。
三、分组集
窗口聚合还支持 GROUPING SETS 语法。分组集允许比标准 GROUP BY 描述的分组操作更复杂的分组操作。行按每个指定的分组集单独分组并为每个组计算聚合就像简单的 GROUP BY 子句一样。
使用 GROUPING SETS 的窗口聚合要求 window_start 和 window_end 列必须位于 GROUP BY 子句中但不能位于 GROUPING SETS 子句中。
Flink SQL SELECT window_start, window_end, supplier_id, SUM(price) as priceFROM TABLE(TUMBLE(TABLE Bid, DESCRIPTOR(bidtime), INTERVAL 10 MINUTES))GROUP BY window_start, window_end, GROUPING SETS ((supplier_id), ());
--------------------------------------------------------
| window_start | window_end | supplier_id | price |
--------------------------------------------------------
| 2020-04-15 08:00 | 2020-04-15 08:10 | (NULL) | 11.00 |
| 2020-04-15 08:00 | 2020-04-15 08:10 | supplier2 | 5.00 |
| 2020-04-15 08:00 | 2020-04-15 08:10 | supplier1 | 6.00 |
| 2020-04-15 08:10 | 2020-04-15 08:20 | (NULL) | 10.00 |
| 2020-04-15 08:10 | 2020-04-15 08:20 | supplier2 | 9.00 |
| 2020-04-15 08:10 | 2020-04-15 08:20 | supplier1 | 1.00 |
--------------------------------------------------------GROUPING SETS 的每个子列表可以指定零个或多个列或表达式并以与直接在 GROUP BY 子句中使用相同的方式进行解释。空分组集意味着所有行都聚合到一个组即使不存在输入行也会输出该组。
对于未出现这些列的分组集对分组列或表达式的引用将替换为结果行中的空值。
四、ROLLUP
ROLLUP 是用于指定常见类型的分组集的简写符号。它表示给定的表达式列表和列表的所有前缀包括空列表。
使用 ROLLUP 进行窗口聚合要求 window_start 和 window_end 列必须位于 GROUP BY 子句中但不能位于 ROLLUP 子句中。
例如以下查询与上面的查询等效。
SELECT window_start, window_end, supplier_id, SUM(price) as price
FROM TABLE(TUMBLE(TABLE Bid, DESCRIPTOR(bidtime), INTERVAL 10 MINUTES))
GROUP BY window_start, window_end, ROLLUP (supplier_id);五、CUBE
CUBE 是用于指定常见类型的分组集的简写符号。它表示给定的列表及其所有可能的子集 - 幂集。
使用 CUBE 的窗口聚合要求 window_start 和 window_end 列必须位于 GROUP BY 子句中但不能位于 CUBE 子句中。
例如以下两个查询是等效的。
SELECT window_start, window_end, item, supplier_id, SUM(price) as priceFROM TABLE(TUMBLE(TABLE Bid, DESCRIPTOR(bidtime), INTERVAL 10 MINUTES))GROUP BY window_start, window_end, CUBE (supplier_id, item);SELECT window_start, window_end, item, supplier_id, SUM(price) as priceFROM TABLE(TUMBLE(TABLE Bid, DESCRIPTOR(bidtime), INTERVAL 10 MINUTES))GROUP BY window_start, window_end, GROUPING SETS ((supplier_id, item),(supplier_id ),( item),( )
)六、选择组窗口开始和结束时间戳
分组窗口的开始和结束时间戳可以通过 window_start 和 window_end 来选定.
七、多级窗口聚合
window_start 和 window_end 列是普通的时间戳字段并不是时间属性。因此它们不能在后续的操作中当做时间属性进行基于时间的操作。 为了传递时间属性需要在 GROUP BY 子句中添加 window_time 列。window_time 是 Windowing TVFs 产生的三列之一它是窗口的时间属性。 window_time 添加到 GROUP BY 子句后就能被选定了。下面的查询可以把它用于后续基于时间的操作比如多级窗口聚合 和 Window TopN。
下面展示了一个多级窗口聚合第一个窗口聚合后把时间属性传递给第二个窗口聚合。
-- 每个supplier_id翻滚5分钟
CREATE VIEW window1 AS
-- 注意内部Window TVF 的窗口开始和窗口结束字段在select 子句中是可选的。但是如果它们出现在子句中则需要为它们起别名以防止名称与外部窗口 TVF 的窗口开始和窗口结束发生冲突。
SELECT window_start as window_5mintumble_start, window_end as window_5mintumble_end, window_time as rowtime, SUM(price) as partial_priceFROM TABLE(TUMBLE(TABLE Bid, DESCRIPTOR(bidtime), INTERVAL 5 MINUTES))GROUP BY supplier_id, window_start, window_end, window_time;-- 在第一个窗口翻滚 10 分钟
SELECT window_start, window_end, SUM(partial_price) as total_priceFROM TABLE(TUMBLE(TABLE window1, DESCRIPTOR(rowtime), INTERVAL 10 MINUTES))GROUP BY window_start, window_end;八、分组窗口聚合
警告分组窗口聚合已经过时。推荐使用更加强大和有效的窗口表值函数聚合。
“窗口表值函数聚合相对于分组窗口聚合有如下优点
包含 性能调优 中提到的所有性能优化。支持标准的 GROUPING SETS 语法。可以在窗口聚合结果上使用 窗口 TopN。等等。
分组窗口聚合定义在 SQL 的 GROUP BY 子句中。和普通的 GROUP BY 子句一样包含分组窗口函数的 GROUP BY 子句的查询会对各组分别计算各产生一个结果行。批处理表和流表上的SQL支持以下分组窗口函数
分组窗口函数
Group Window FunctionDescriptionTUMBLE(time_attr, interval)定义一个滚动时间窗口。它把数据分配到连续且不重叠的固定时间区间interval,例如一个5分钟的滚动窗口以5分钟为间隔对数据进行分组。滚动窗口可以被定义在事件时间流 批或者处理时间流上。HOP(time_attr, interval, interval)定义一个滑动时间窗口它有窗口大小第二个 interval 参数和滑动间隔第一个 interval 参数两个参数。如果滑动间隔小于窗口大小窗口会产生重叠。所以数据可以被指定到多个窗口。例如一个15分钟大小和5分钟滑动间隔的滑动窗口将每一行分配给3个15分钟大小的不同窗口这些窗口以5分钟的间隔计算。滑动窗口可以被定义在事件时间流 批或者处理时间流上。SESSION(time_attr, interval)定义一个会话时间窗口。会话时间窗口没有固定的时间区间但其边界是通过不活动的时间 interval 定义的即一个会话窗口会在指定的时长内没有事件出现时关闭。例如一个30分钟间隔的会话窗口收到一条数据时如果之前已经30分钟不活动了否则这条数据会被分配到已经存在的窗口中它会开启一个新窗口如果30分钟之内没有新数据到来就会关闭。会话窗口可以被定义在事件时间流 批) 或者处理时间流上。
九、时间属性
在流处理模式分组窗口函数的 time_attr 属性必须是一个有效的处理或事件时间。在批处理模式分组窗口函数的 time_attr 参数必须是一个 TIMESTAMP 类型的属性。
十、选取分组窗口开始和结束时间戳
分组窗口的开始和结束时间戳以及时间属性也可以通过下列辅助函数的方式获取到
辅助函数描述TUMBLE_START(time_attr, interval)、HOP_START(time_attr, interval, interval)、SESSION_START(time_attr, interval)返回相应的滚动滑动或会话窗口的下限的时间戳inclusive即窗口开始时间。TUMBLE_END(time_attr, interval)、HOP_END(time_attr, interval, interval)、SESSION_END(time_attr, interval)返回相应滚动窗口跳跃窗口或会话窗口的上限的时间戳exclusive即窗口结束时间。注意: 上限时间戳exlusive不能作为 rowtime attribute 用于后续基于时间的操作例如interval joins 和 group window 或 over window aggregations。TUMBLE_ROWTIME(time_attr, interval)、HOP_ROWTIME(time_attr, interval, interval)、SESSION_ROWTIME(time_attr, interval)返回相应滚动窗口跳跃窗口或会话窗口的上限的时间戳inclusive即窗口事件时间或窗口处理时间。返回的值是 rowtime attribute可以用于后续基于时间的操作比如interval joins 和 group window 或 over window aggregations。TUMBLE_PROCTIME(time_attr, interval)、HOP_PROCTIME(time_attr, interval, interval)、SESSION_PROCTIME(time_attr, interval)返回的值是 proctime attribute可以用于后续基于时间的操作比如 interval joins 和 group window 或 over window aggregations。
注意 辅助函数的参数必须和 GROUP BY 子句中的分组窗口函数一致。
下面的例子展示了在流式表上如何使用分组窗口 SQL 查询
CREATE TABLE Orders (user BIGINT,product STRING,amount INT,order_time TIMESTAMP(3),WATERMARK FOR order_time AS order_time - INTERVAL 1 MINUTE
) WITH (...);SELECTuser,TUMBLE_START(order_time, INTERVAL 1 DAY) AS wStart,SUM(amount) FROM Orders
GROUP BYTUMBLE(order_time, INTERVAL 1 DAY),user
