实时计算Flink版：窗口函数

Flink SQL支持的窗口聚合主要是两种：Window聚合和Over聚合。本文档主要为您介绍Window聚合。Window聚合支持Event Time和Processing Time两种时间属性定义窗口。每种时间属性类型支持三种窗口类型：滚动窗口（TUMBLE）、滑动窗口（HOP）和会话窗口（SESSION）。

Flink SQL支持Event Time和Processing Time两种时间属性，时间属性详情请参见时间属性。Flink可以基于这两种时间属性对数据进行窗口聚合。基于这两种时间属性开窗的区别如下：

• Event Time：您提供的事件时间，通常是数据的最原始的创建时间。

  系统会根据数据的Event Time生成的Watermark来进行关窗。只有当Watermark大于关窗时间，才会触发窗口的结束，窗口结束才会输出结果。如果一直没有触发窗口结束的数据流入Flink，则该窗口就无法输出数据。单个subtask的Watermark是递增的，多个subtask或者多个源表的Watermark取最小值。

  重要

  • 如果源表中存在一条未来的乱序数据或者某个subtask或上游源表的某个分区中没有数据，则可能会无法触发窗口结束，从而导致结果数据异常。因此您需要根据数据乱序的程度设置合理的offset大小，并保证所有subtask和上游源表的所有分区中都有数据。如果某个subtask或上游源表的某个分区中没有数据，导致Watermark无法推进，窗口无法及时结束，则可以在更多Flink配置中添加table.exec.source.idle-timeout: 10s来触发窗口结束。该参数含义详情请参见Configuration。

  • 数据经过GroupBy、双流JOIN或OVER窗口节点后，会导致Watermark属性丢失，无法再使用Event Time进行开窗。

• Processing Time：对事件进行处理的本地系统时间。

  Processing Time是Flink系统产生的，不包含在用户的原始数据中。因此需要您显式定义一个Processing Time列。

  说明

  因为Processing Time容易受到事件到达Flink系统的速度及Flink内部处理数据顺序的影响，所以每次回溯数据的结果可能不一致。

示例逻辑为：基于1分钟的滚动窗口聚合结果，进行1小时的滚动窗口聚合，可以满足您的多维度开窗需求。

CREATE TEMPORARY TABLE user_clicks(
  username varchar,
  click_url varchar,
  eventtime varchar,                                                        
  ts AS TO_TIMESTAMP(eventtime),
  WATERMARK FOR ts AS ts - INTERVAL '2' SECOND  --为Rowtime定义Watermark。
) with (
  'connector'='sls',
  ...
);

CREATE TEMPORARY TABLE tumble_output(
  window_start TIMESTAMP,
  window_end TIMESTAMP,
  username VARCHAR,
  clicks BIGINT
) with (
  'connector'='datahub'       --目前SLS只支持输出VARCHAR类型的DDL，所以使用DataHub存储。
  ...
);

CREATE TEMPORARY VIEW one_minute_window_output AS 
SELECT 
  TUMBLE_ROWTIME(ts, INTERVAL '1' MINUTE) as rowtime,  --使用TUMBLE_ROWTIME作为二级Window的聚合时间。
  username, 
  COUNT(click_url) as cnt
FROM user_clicks
GROUP BY TUMBLE(ts, INTERVAL '1' MINUTE),username;

BEGIN statement set;
INSERT INTO tumble_output
SELECT
  TUMBLE_START(rowtime, INTERVAL '1' HOUR),
  TUMBLE_END(rowtime, INTERVAL '1' HOUR),
  username,
  SUM(cnt)
FROM one_minute_window_output
GROUP BY TUMBLE(rowtime, INTERVAL '1' HOUR), username;
END;