OVER窗口函数使用语法及实例 - 实时计算Flink版

OVER窗口（OVER Window）是传统数据库的标准开窗，不同于Group By Window，OVER窗口中每1个元素都对应1个窗口。OVER窗口可以按照实际元素的行或实际的元素值（时间戳值）确定窗口，因此流数据元素可能分布在多个窗口中。

在应用OVER窗口的流式数据中，每1个元素都对应1个OVER窗口。每1个元素都触发1次数据计算，每个触发计算的元素所确定的行，都是该元素所在窗口的最后1行。在实时计算的底层实现中，OVER窗口的数据进行全局统一管理（数据只存储1份），逻辑上为每1个元素维护1个OVER窗口，为每1个元素进行窗口计算，完成计算后会清除过期的数据。详情请参见Over Aggregation。

语法

SELECT
    agg1(col1) OVER (definition1) AS colName,
    ...
    aggN(colN) OVER (definition1) AS colNameN
FROM Tab1;

agg1(col1)：按照GROUP BY指定col1列对输入数据进行聚合计算。
OVER (definition1)：OVER窗口定义。
AS colName：别名。

说明

agg1到aggN所对应的OVER definition1必须相同。
外层SQL可以通过AS的别名查询数据。

类型

Flink SQL中对OVER窗口的定义遵循标准SQL的定义语法，传统OVER窗口没有对其进行更细粒度的窗口类型命名划分。按照计算行的定义方式，OVER Window可以分为以下两类：

ROWS OVER Window：每1行元素都被视为新的计算行，即每1行都是一个新的窗口。
RANGE OVER Window：具有相同时间值的所有元素行视为同一计算行，即具有相同时间值的所有行都是同一个窗口。

属性

正交属性	说明	proctime	eventtime
ROWS OVER Window	按照实际元素的行确定窗口。	支持	支持
RANGE OVER Window	按照实际的元素值（时间戳值）确定窗口。	支持	支持

Rows OVER Window语义

窗口数据
ROWS OVER Window的每个元素都确定一个窗口。

窗口语法

SELECT
    agg1(col1) OVER(
     [PARTITION BY (value_expression1,..., value_expressionN)]
     ORDER BY timeCol
     ROWS 
     BETWEEN (UNBOUNDED | rowCount) PRECEDING AND CURRENT ROW) AS colName, ...
FROM Tab1;

value_expression：分区值表达式。
timeCol：元素排序的时间字段。
rowCount：定义根据当前行开始向前追溯几行元素。

案例

以Bounded ROWS OVER Window场景为例。假设有一张商品上架表，包含商品ID、商品类型、商品上架时间、商品价格数据。要求输出在当前商品上架之前同类的3个商品中的最高价格。

测试表tmall_item数据

itemid(VARCHAR)	itemtype(VARCHAR)	eventtime(VARCHAR)	price(DOUBLE)
ITEM001	Electronic	`2024-11-11 10:01:00`	20
ITEM002	Electronic	`2024-11-11 10:02:00`	50
ITEM003	Electronic	`2024-11-11 10:03:00`	30
ITEM004	Electronic	`2024-11-11 10:03:00`	60
ITEM005	Electronic	`2024-11-11 10:05:00`	40
ITEM006	Electronic	`2024-11-11 10:06:00`	20
ITEM007	Electronic	`2024-11-11 10:07:00`	70
ITEM008	Clothes	`2024-11-11 10:08:00`	20

测试代码

CREATE TEMPORARY TABLE tmall_item(
  itemid VARCHAR,
  itemtype VARCHAR,
  eventtime varchar,                            
  onselltime AS TO_TIMESTAMP(eventtime),
  price DOUBLE,
  WATERMARK FOR onselltime AS onselltime - INTERVAL '2' SECOND  -- 为Rowtime定义Watermark
) WITH (
  'connector' = 'kafka',
  'topic' = '<yourTopic>',
  'properties.bootstrap.servers' = '<brokers>',
  'scan.startup.mode' = 'earliest-offset',
  'format' = 'csv'
);

SELECT
    itemid,
    itemtype,
    onselltime,
    price,  
    MAX(price) OVER (
        PARTITION BY itemtype 
        ORDER BY onselltime 
        ROWS BETWEEN 2 preceding AND CURRENT ROW) AS maxprice
FROM tmall_item;

测试结果

itemid	itemtype	onselltime	price	maxprice
ITEM001	Electronic	`2024-11-11 10:01:00`	20	20
ITEM002	Electronic	`2024-11-11 10:02:00`	50	50
ITEM003	Electronic	`2024-11-11 10:03:00`	30	50
ITEM004	Electronic	`2024-11-11 10:03:00`	60	60
ITEM005	Electronic	`2024-11-11 10:05:00`	40	60
ITEM006	Electronic	`2024-11-11 10:06:00`	20	60
ITEM007	Electronic	`2024-11-11 10:07:00`	70	70
ITEM008	Clothes	`2024-11-11 10:08:00`	20	20

RANGE OVER Window语义

窗口数据
RANGE OVER Window所有具有共同元素值（元素时间戳）的元素行确定一个窗口。

窗口语法

SELECT
    agg1(col1) OVER(
     [PARTITION BY (value_expression1,..., value_expressionN)]
     ORDER BY timeCol
     RANGE 
     BETWEEN (UNBOUNDED | timeInterval) PRECEDING AND CURRENT ROW) AS colName,
...
FROM Tab1;

value_expression：进行分区的字表达式。
timeCol：元素排序的时间字段。
timeInterval：定义根据当前行开始向前追溯指定时间的元素行。

案例

Bounded RANGE OVER Window场景示例：假设一张商品上架表，包含有商品ID、商品类型、商品上架时间、商品价格数据。需要求比当前商品上架时间早2分钟的同类商品中的最高价格。

测试表tmall_item数据

itemid(VARCHAR)	itemtype(VARCHAR)	eventtime(VARCHAR)	price(DOUBLE)
ITEM001	Electronic	`2024-11-11 10:01:00`	20
ITEM002	Electronic	`2024-11-11 10:02:00`	50
ITEM003	Electronic	`2024-11-11 10:03:00`	30
ITEM004	Electronic	`2024-11-11 10:03:00`	60
ITEM005	Electronic	`2024-11-11 10:05:00`	40
ITEM006	Electronic	`2024-11-11 10:06:00`	20
ITEM007	Electronic	`2024-11-11 10:07:00`	70
ITEM008	Clothes	`2024-11-11 10:08:00`	20

测试代码

CREATE TEMPORARY TABLE tmall_item(
  itemid VARCHAR,
  itemtype VARCHAR,
  eventtime varchar,                            
  onselltime AS TO_TIMESTAMP(eventtime),
  price DOUBLE,
  WATERMARK FOR onselltime AS onselltime - INTERVAL '2' SECOND  -- 为Rowtime定义Watermark
) WITH (
  'connector' = 'kafka',
  'topic' = '<yourTopic>',
  'properties.bootstrap.servers' = '<brokers>',
  'scan.startup.mode' = 'earliest-offset',
  'format' = 'csv'
);

SELECT  
    itemid,
    itemtype, 
    onselltime, 
    price,  
    MAX(price) OVER (
        PARTITION BY itemtype 
        ORDER BY onselltime 
        RANGE BETWEEN INTERVAL '2' MINUTE preceding AND CURRENT ROW) AS maxprice
FROM tmall_item;

测试结果

itemid	itemtype	onselltime	price	maxprice
ITEM001	Electronic	`2024-11-11 10:01:00`	20	20
ITEM002	Electronic	`2024-11-11 10:02:00`	50	50
ITEM003	Electronic	`2024-11-11 10:03:00`	30	50
ITEM004	Electronic	`2024-11-11 10:03:00`	60	60
ITEM005	Electronic	`2024-11-11 10:05:00`	40	60
ITEM006	Electronic	`2024-11-11 10:06:00`	20	40
ITEM007	Electronic	`2024-11-11 10:07:00`	70	70
ITEM008	Clothes	`2024-11-11 10:08:00`	20	20