阿里云实时计算OVER窗口

语法

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

说明

• agg1到aggN所对应的OVER definition1必须相同。
• AS的别名可供外层SQL进行查询。

类型

Flink SQL中对OVER Window的定义遵循标准SQL的定义语法，传统OVER Window没有对其进行更细粒度的窗口类型命名划分。本节为了方便您理解OVER Window的语义，将OVER Window按照计算行的定义方式划分为如下两类。

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

属性

• rows：按照实际元素的行进行确定窗口。
• range：按照实际的元素值（时间戳值）进行确定窗口。

Rows OVER Window语义

• 窗口数据

  ROWS OVER Window的每个元素都确定一个窗口。ROWS OVER Window也有Unbounded和Bounded的两种情况。

  Unbounded ROWS OVER Window数据如下图所示。

  

  说明 上图所示窗口user1的w7和w8，user2的窗口w3和w4，虽然元素都是同一时刻到达，但是它们仍然是在不同的窗口，这一点有别于RANGE OVER Window。

  Bounded ROWS OVER Window数据以3个元素（2 PRECEDING）的窗口为例，如下图所示。

  

  说明 上图所示窗口user1的w5和w6，user2的窗口w2和w3，虽然有元素都是同一时刻到达，但是它们仍然是在不同的窗口，这一点有别于RANGE 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个商品中的最高价格。

  测试数据

  商品ID	商品类型	上架时间	销售价格
  ITEM001	Electronic	2017-11-11 10:01:00	20
  ITEM002	Electronic	2017-11-11 10:02:00	50
  ITEM003	Electronic	2017-11-11 10:03:00	30
  ITEM004	Electronic	2017-11-11 10:03:00	60
  ITEM005	Electronic	2017-11-11 10:05:00	40
  ITEM006	Electronic	2017-11-11 10:06:00	20
  ITEM007	Electronic	2017-11-11 10:07:00	70
  ITEM008	Clothes	2017-11-11 10:08:00	20

  测试代码

  CREATE TABLE tmall_item(
     itemID VARCHAR,
     itemType VARCHAR,
     onSellTime TIMESTAMP,
     price DOUBLE,
     WATERMARK onSellTime FOR onSellTime as withOffset(onSellTime, 0) 
  ) 
  WITH (
    type = 'sls',
     ...
  ) ;
  
  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	2017-11-11 10:01:00	20	20
  ITEM002	Electronic	2017-11-11 10:02:00	50	50
  ITEM003	Electronic	2017-11-11 10:03:00	30	50
  ITEM004	Electronic	2017-11-11 10:03:00	60	60
  ITEM005	Electronic	2017-11-11 10:05:00	40	60
  ITEM006	Electronic	2017-11-11 10:06:00	20	60
  ITEM007	Electronic	2017-11-11 10:07:00	70	70
  ITEM008	Clothes	2017-11-11 10:08:00	20	20