Window functions

Flink SQL supports window aggregates and over aggregates. This topic covers window aggregates. Window aggregates use two time attributes — event time and processing time — and support the TUMBLE, HOP, and SESSION window functions.

Flink SQL supports two time attributes: event time and processing time. The windowing behavior differs based on which attribute you use.

• Event time: typically the timestamp embedded in a record.

  A window closes when the watermark exceeds the window end time. Output is generated only when closing data arrives. For a single subtask, the watermark increments monotonically. With multiple subtasks or source tables, Flink uses the minimum watermark value.

  Important

  • If out-of-order records exist or a subtask or partition has no data, the watermark cannot advance and the window may not close. To resolve this, specify a watermark offset for out-of-order data, and ensure all subtasks and partitions have data flowing in. If a partition is idle, add table.exec.source.idle-timeout: 10s to the Other Configuration field in Parameters on the Configuration tab of the Deployments page. See Configuration for parameter details.

  • After data is processed by using GROUP BY, JOIN operations on two data streams, or OVER window nodes, the watermark property is lost and the event time can no longer be used for windowing.

• Processing time: the system clock time when Flink processes an event.

  Processing time is generated by Flink and does not exist in your raw data, so you must explicitly define a processing time column.

  Note

  Because processing time depends on event arrival speed and processing order, backtrack results may differ between runs.

The following example cascades a 1-hour tumbling window on top of a 1-minute tumbling window:

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   -- Define a watermark for rowtime. 
) with (
  'connector'='sls',
  ...
);

CREATE TEMPORARY TABLE tumble_output(
  window_start TIMESTAMP,
  window_end TIMESTAMP,
  username VARCHAR,
  clicks BIGINT
) with (
  'connector'='datahub'        -- Simple Log Service allows you to export only VARCHAR-type DDL statements. Therefore, DataHub is used to store data. 
  ...
);

CREATE TEMPORARY VIEW one_minute_window_output AS 
SELECT 
  TUMBLE_ROWTIME(ts, INTERVAL '1' MINUTE) as rowtime,  -- Use TUMBLE_ROWTIME as the aggregation time of the level-two 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;