FlinkSql系列3之窗口聚合和GROUP BY的相关区别

前言

使用window然后按照窗口可以进行聚合，但是其实使用group by单纯也是可以实现的，但是两者还是有很大区别的。

一、window group aggregation & group by key

CREATE TABLE source_table(
--维度数据
`city_id` STRING,
--用户id
`user_id` BIGINT,
--金额
`price` BIGINT,
--事件时间戳
`rowtime` AS CAST(CURRENT_TIMESTAMP AS timestamp(3)),
--watermark设置
WATERMARK FOR rowtime AS rowtime - INTERVAL '3' second
)WITH (
 'connector' = 'datagen',
 'rows-per-second' = '5',
 'fields.city_id.length' = '1',
 'fields.user_id.min' = '1',
 'fields.user_id.max' = '50',
 'fields.price.min' = '1',
 'fields.price.max' = '50'
);
 
 
CREATE TABLE sink_table(
--维度数据
`city_id` STRING,
--pv pagevisit
`pv` BIGINT,
--总金额
`sum_price` BIGINT,
--最大金额
`max_price` BIGINT,
--最小金额
`min_price` BIGINT,
--uv 去重
`uv` BIGINT,
--窗口开始时间
`winodwstart` BIGINT,
--窗口结束时间
`windowend` BIGINT
) WITH (
'connector'='print'
)

INSERT INTO sink_table
SELECT
city_id,
COUNT(user_id) as pv,
SUM(price) as sum_price,
MAX(price) as max_price,
MIN(price) as min_price,
COUNT(DISTINCT user_id) as uv,
UNIX_TIMESTAMP(CAST(TUMBLE_START(rowtime,INTERVAL '1' MINUTE) AS STRING)) * 1000 as window_start
FROM source_table
GROUP BY 
city_id,
TUMBLE(rowtime,INTERVAL '1' MINUTE)

我们使用GROUP BY改写这个，因为我们开的是一分钟的滚动窗口，对于rowtime来说，也就是比如说1.30.10 和1.30.20这种都属于1.30-1.31，所以我们可以把rowtime换算为分钟级别，这样对于1.30.10 和1.30.20都会变成1.30了，自然就可以group by 了。

INSERT INTO sink_table
SELECT
city_id,
COUNT(user_id) as pv,
SUM(price) as sum_price,
MAX(price) as max_price,
MIN(price) as min_price,
COUNT(DISTINCT user_id) as uv,
CAST(UNIX_TIMESTAMP(CAST(rowtime AS STRING)) / 60 AS BIGINT) as window_start
FROM source_table
GROUP BY
city_id,
CAST((UNIX_TIMESTAMP(CAST(rowtime AS STRING)) / 60 ) AS BIGINT)

总结

窗⼝聚合和 Group by 聚合的差异在于：
本质区别：窗⼝聚合是具有时间语义的，其本质是想实现窗⼝结束输出结果之后，后续有迟到的数据也 不会对原有的结果发⽣更改了，即输出结果值是定值（不考虑 allowLateness）。⽽ Group by 聚合是没有 时间语义的，不管数据迟到多⻓时间，只要数据来了，就把上⼀次的输出的结果数据撤回，然后把计算好的 新的结果数据发出
运⾏层⾯：窗⼝聚合是和 时间 绑定的，窗⼝聚合其中窗⼝的计算结果触发都是由时间（Watermark） 推动的。Group by 聚合完全由数据推动触发计算，新来⼀条数据去根据这条数据进⾏计算出结果发出；由 此可⻅两者的实现⽅式也⼤为不同。