FlinkSql中的聚合查询

SELECT user, COUNT(url) as cnt FROM EventTable GROUP BY user

开窗聚合

        开窗函数的聚合与之前两种聚合有本质的不同：分组聚合、窗口 TVF聚合都是“多对一”的关系，将数据分组之后每组只会得到一个聚合结果；而开窗函数是对每行都要做一次开窗聚合，因此聚合之后表中的行数不会有任何减少，是一个“多对多”的关系。Flink SQL 中的开窗函数也是通过 OVER 子句来实现的，这里 OVER 关键字前面是一个聚合函数，它会应用在后面 OVER 定义的窗口上。在 OVER子句中主要有以下几个部分：

• PARTITION BY（可选）：用来指定分区的键（key），类似于 GROUP BY 的分组，这部分是可选的；
• ORDER BY：明确地指出数据基于那个字段排序（目前只支持按照时间属性的升序排列，所以这里 ORDER BY 后面的字段必须是定义好的时间属性）
• 开窗范围：要扩展多少行来做聚合。这个范围是由 BETWEEN <下界> AND <上界> 来定义的，也就是“从下界到上界”的范围。目前支持的上界只能是 CURRENT ROW，也就是定义一个“从之前某一行到当前行”的范围。

开窗范围

        开窗选择的范围可以基于时间，也可以基于数据的数量。所以开窗范围还应该在两种模式之间做出选择：范围间隔（RANGE intervals）和行间隔（ROW intervals）。

范围间隔

        范围间隔以 RANGE 为前缀，就是基于 ORDER BY 指定的时间字段去选取一个范围，一般就是当前行时间戳之前的一段时间。例如开窗范围选择当前行之前 1 小时的数据：

RANGE BETWEEN INTERVAL '1' HOUR PRECEDING AND CURRENT ROW

行间隔

        行间隔以 ROWS 为前缀，就是直接确定要选多少行，由当前行出发向前选取就可以了。例如开窗范围选择当前行之前的 5 行数据（最终聚合会包括当前行，所以一共 6 条数据）：       

ROWS BETWEEN 5 PRECEDING AND CURRENT ROW

使用场景：统计每个用户截止当前事件事件一小时内的uv

SELECT user, ts,
 COUNT(url) OVER (
 PARTITION BY user
 ORDER BY ts
 RANGE BETWEEN INTERVAL '1' HOUR PRECEDING AND CURRENT ROW
 ) AS cnt
FROM EventTable

WINDOW 子句

        在 SQL 中，也可以用 WINDOW 子句来在 SELECT 外部单独定义一个 OVER 窗口，来实现窗口的复用。

SELECT user, ts,
 COUNT(url) OVER w AS cnt,
 MAX(CHAR_LENGTH(url)) OVER w AS max_url
FROM EventTable
WINDOW w AS (
 PARTITION BY user
 ORDER BY ts
 ROWS BETWEEN 2 PRECEDING AND CURRENT ROW)

窗口聚合

SELECT  
 user,  
 window_end AS endT,  
 COUNT(url) AS cnt  
 FROM TABLE(  
	CUMULATE( TABLE EventTable,   // 定义累积窗口
	DESCRIPTOR(ts),  
	INTERVAL '1' HOUR,  
	INTERVAL '1' DAY))  
 GROUP BY user, window_start, window_end 

聚合实例

统计每小时内有最多访问行为的用户，取前两名，相当于是一个每小时活跃用户的查询（窗口topN）

SELECT *  
FROM 
	( 
	SELECT *,  
	ROW_NUMBER() OVER (PARTITION BY window_start, window_end ORDER BY cnt desc ) AS row_num  
	FROM 
		(  
		--开一个滚动窗口 
		SELECT window_start, window_end, user, COUNT(url) as cnt  
		FROM TABLE (TUMBLE( TABLE EventTable, DESCRIPTOR(ts), INTERVAL '1' HOUR )) 
		GROUP BY window_start, window_end, user
		) a
	) a  
WHERE row_num <= 2
;