架构 大家好，聚合我是操作老羊，今天我们来学习 Flink SQL 中的聚合· Over 聚合操作。
那这里我们拿 Over 聚合 与 窗口聚合 做一个对比，聚合其之间的操作最大不同之处在于： 窗口聚合：不在 group by 中的字段，不能直接在 select 中拿到 Over 聚合：能够保留原始字段 注意：其实在生产环境中，聚合Over 聚合的操作使用场景还是比较少的。在 Hive 中也有相同的聚合聚合，但是操作小伙伴萌可以想想你在离线数仓经常使用嘛？ 应用场景：计算最近一段滑动窗口的聚合结果数据。IT技术网际案例：查询每个产品最近一小时订单的聚合金额总和：复制 SELECT order_id, order_time, amount, SUM(amount) OVER ( PARTITION BYproduct ORDER BYorder_time RANGE BETWEEN INTERVAL 1 HOUR PRECEDING ANDCURRENT ROW ) ASone_hour_prod_amount_sum FROM Orders1.2.3.4.5.6.7.8.Over 聚合的语法总结如下： 复制SELECT agg_func(agg_col) OVER ( [PARTITION BY col1[, col2, ...]] ORDER BYtime_col range_definition),... FROM ...1.2.3.4.5.6.7.其中： ORDER BY：必须是时间戳列（事件时间、处理时间）PARTITION BY：标识了聚合窗口的操作聚合粒度，如上述案例是聚合按照 product 进行聚合range_definition：这个标识聚合窗口的聚合数据范围，在 Flink 中有两种指定数据范围的操作方式。第一种为按照行数聚合，聚合第二种为按照时间区间聚合。如下案例所示：a. 时间区间聚合： 按照时间区间聚合就是时间区间的一个滑动窗口，比如下面案例 1 小时的区间，最新输出的一条数据的 sum 聚合结果就是高防服务器最近一小时数据的 amount 之和。 复制CREATE TABLE source_table ( 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.001SECOND ) 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);CREATE TABLE sink_table ( product BIGINT, order_time TIMESTAMP(3), amount BIGINT, one_hour_prod_amount_sum BIGINT) WITH ( connector = print);INSERT INTOsink_table SELECT product, order_time, amount, SUM(amount) OVER ( PARTITION BYproduct ORDER BYorder_time -- 标识统计范围是一个 product 的最近 1 小时的数据 RANGE BETWEEN INTERVAL 1 HOUR PRECEDING ANDCURRENT ROW ) ASone_hour_prod_amount_sum FROM source_table1.2.3.4.5.6.7.8.9.10.11.12.13.14.15.16.17.18.19.20.21.22.23.24.25.26.27.28.29.30.31.32.33.34.35.结果如下： 复制+I[2, 2021-12-24T22:08:26.583, 7, 73]+I[2, 2021-12-24T22:08:27.583, 7, 80]+I[2, 2021-12-24T22:08:28.583, 4, 84]+I[2, 2021-12-24T22:08:29.584, 7, 91]+I[2, 2021-12-24T22:08:30.583, 8, 99]+I[1, 2021-12-24T22:08:31.583, 9, 138]+I[2, 2021-12-24T22:08:32.584, 6, 105]+I[1, 2021-12-24T22:08:33.584, 7, 145]1.2.3.4.5.6.7.8.b. 行数聚合： 按照行数聚合就是数据行数的一个滑动窗口，比如下面案例，最新输出的一条数据的 sum 聚合结果就是最近 5 行数据的 amount 之和。 复制CREATE TABLE source_table ( 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.001SECOND ) 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);CREATE TABLE sink_table ( product BIGINT, order_time TIMESTAMP(3), amount BIGINT, one_hour_prod_amount_sum BIGINT) WITH ( connector = print);INSERT INTOsink_table SELECT product, order_time, amount, SUM(amount) OVER ( PARTITION BYproduct ORDER BYorder_time -- 标识统计范围是一个 product 的最近 5 行数据 ROWS BETWEEN 5 PRECEDING ANDCURRENT ROW ) ASone_hour_prod_amount_sum FROM source_table1.2.3.4.5.6.7.8.9.10.11.12.13.14.15.16.17.18.19.20.21.22.23.24.25.26.27.28.29.30.31.32.33.34.35.预跑结果如下： 复制+I[2, 2021-12-24T22:18:19.147, 1, 9]+I[1, 2021-12-24T22:18:20.147, 2, 11]+I[1, 2021-12-24T22:18:21.147, 2, 12]+I[1, 2021-12-24T22:18:22.147, 2, 12]+I[1, 2021-12-24T22:18:23.148, 2, 12]+I[1, 2021-12-24T22:18:24.147, 1, 11]+I[1, 2021-12-24T22:18:25.146, 1, 10]+I[1, 2021-12-24T22:18:26.147, 1, 9]+I[2, 2021-12-24T22:18:27.145, 2, 11]+I[2, 2021-12-24T22:18:28.148, 1, 10]+I[2, 2021-12-24T22:18:29.145, 2, 10]1.2.3.4.5.6.7.8.9.10.11.当然，如果你在一个 SELECT 中有多个聚合窗口的聚合方式，Flink SQL 支持了一种简化写法，如下案例： 复制SELECT order_id, order_time, amount, SUM(amount) OVER w AS sum_amount, AVG(amount) OVER w ASavg_amount FROMOrders -- 使用下面子句，香港云服务器定义 Over WindowWINDOW w AS ( PARTITION BYproduct ORDER BYorder_time RANGE BETWEEN INTERVAL 1 HOUR PRECEDING AND CURRENT ROW)1.2.3.4.5.6.7.8.9. |
