SESSION Window Functions Overview for Irregular Event Streams - Realtime Compute for Apache Flink

SESSION groups stream elements by session activity, creating variable-size, non-overlapping windows. A session window closes when no elements arrive within the inactivity period defined by the session gap. Elements arriving before the gap expires are merged into the same session; elements arriving after the gap open a new session window.

For example, with a 10-minute session gap: if the interval between two events from the same user is less than 10 minutes, both events belong to the same session window. If no event arrives within 10 minutes after the latest event, the window closes and is sent downstream. Subsequent events open a new session window.

Syntax

Use SESSION in a GROUP BY clause to define a session window.

SESSION(<time-attr>, <gap-interval>)

Parameters

Parameter	Description	Example
`time-attr`	A valid time attribute field in the stream. Specifies whether to use processing time or event time. For details, see Time attributes.	-
`gap-interval`	The session gap: the period of inactivity after which the session window closes. Format: `INTERVAL 'num' timeUnit`.	`INTERVAL '10' SECOND` (10-second gap)

Window identifier functions

Window identifier functions return the start time, end time, or time attribute of a session window. Use them in a SELECT clause to retrieve window boundaries.

Function	Return type	Description
`SESSION_START(<time-attr>, <gap-interval>)`	TIMESTAMP	Returns the start time of the window, inclusive. For example, for the window `[00:10, 00:15]`, returns `00:10`. The value equals the time of the first record in the session.
`SESSION_END(<time-attr>, <gap-interval>)`	TIMESTAMP	Returns the end time of the window, inclusive. For example, for the window `[00:00, 00:15]`, returns `00:15`. The value equals the time of the last record plus the session gap.
`SESSION_ROWTIME(<time-attr>, <gap-interval>)`	TIMESTAMP (rowtime attribute)	Returns the end time of the window, exclusive. For example, for the window `(00:00, 00:15)`, returns `00:14:59.999`. The return value is a rowtime attribute that supports subsequent time-based operations. Applies to event time windows only. For details, see Cascading windows.
`SESSION_PROCTIME(<time-attr>, <gap-interval>)`	TIMESTAMP (processing time attribute)	Returns the end time of the window, exclusive. For example, for the window `(00:00, 00:15)`, returns `00:14:59.999`. The return value is a processing time attribute that supports subsequent time-based operations. Applies to processing time windows only. For details, see Cascading windows.

Example

This example counts clicks per user per active session, using a 30-second session gap and event time.

Test data (user_clicks table)

username (VARCHAR)	click_url (VARCHAR)	eventtime (VARCHAR)
Jark	`http://taobao.com/xxx`	`2024-10-10 10:00:00.0`
Jark	`http://taobao.com/xxx`	`2024-10-10 10:00:10.0`
Jark	`http://taobao.com/xxx`	`2024-10-10 10:00:49.0`
Jark	`http://taobao.com/xxx`	`2024-10-10 10:01:05.0`
Jark	`http://taobao.com/xxx`	`2024-10-10 10:01:58.0`
Timo	`http://taobao.com/xxx`	`2024-10-10 10:02:10.0`

SQL statements

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 the rowtime.
) WITH (
  'connector' = 'kafka',
  'topic' = '<yourTopic>',
  'properties.bootstrap.servers' = '<brokers>',
  'scan.startup.mode' = 'earliest-offset',
  'format' = 'csv'
);

CREATE TEMPORARY TABLE session_output(
  window_start TIMESTAMP,
  window_end TIMESTAMP,
  username VARCHAR,
  clicks BIGINT
) WITH (
  'connector'='print',
  'logger'='true'
);

INSERT INTO session_output
SELECT
  SESSION_START(ts, INTERVAL '30' SECOND),
  SESSION_END(ts, INTERVAL '30' SECOND),
  username,
  COUNT(click_url)
FROM user_clicks
GROUP BY SESSION(ts, INTERVAL '30' SECOND), username;

Results

window_start (TIMESTAMP)	window_end (TIMESTAMP)	username (VARCHAR)	clicks (BIGINT)
`2024-10-10 10:00:00.0`	`2024-10-10 10:00:40.0`	Jark	2
`2024-10-10 10:00:49.0`	`2024-10-10 10:01:35.0`	Jark	2
`2024-10-10 10:01:58.0`	`2024-10-10 10:02:28.0`	Jark	1
`2024-10-10 10:02:10.0`	`2024-10-10 10:02:40.0`	Timo	1