SESSION - Realtime Compute for Apache Flink: Central de Documentação do Alibaba Cloud

A função SESSION agrupa elementos de stream por atividade de sessão e cria janelas de tamanho variável sem sobreposição. A janela de sessão fecha quando nenhum elemento chega durante o período de inatividade definido pelo intervalo da sessão. Elementos recebidos antes do fim desse intervalo são mesclados na mesma sessão; elementos recebidos após o intervalo abrem uma nova janela de sessão.

Por exemplo, com um intervalo de sessão de 10 minutos: se o tempo entre dois eventos do mesmo usuário for inferior a 10 minutos, ambos pertencem à mesma janela de sessão. Se nenhum evento ocorrer nos 10 minutos seguintes ao último evento, a janela fecha e é enviada para downstream. Eventos subsequentes iniciam uma nova janela de sessão.

Sintaxe

Use SESSION em uma cláusula GROUP BY para definir uma janela de sessão.

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

Parâmetros

Parâmetro	Descrição	Exemplo
`time-attr`	Campo de atributo de tempo válido no stream. Define se deve ser usado o tempo de processamento ou o tempo do evento. Para mais detalhes, consulte Time attributes.	-
`gap-interval`	Intervalo da sessão: período de inatividade após o qual a janela de sessão fecha. Formato: `INTERVAL 'num' timeUnit`.	`INTERVAL '10' SECOND` (intervalo de 10 segundos)

Funções identificadoras de janela

As funções identificadoras de janela retornam o horário de início, o horário de término ou o atributo de tempo de uma janela de sessão. Use-as em uma cláusula SELECT para obter os limites da janela.

Função	Tipo de retorno	Descrição
`SESSION_START(<time-attr>, <gap-interval>)`	TIMESTAMP	Retorna o horário de início da janela (inclusivo). Por exemplo, para a janela `[00:10, 00:15]`, o retorno é `00:10`. Esse valor corresponde ao horário do primeiro registro da sessão.
`SESSION_END(<time-attr>, <gap-interval>)`	TIMESTAMP	Retorna o horário de término da janela (inclusivo). Por exemplo, para a janela `[00:00, 00:15]`, o retorno é `00:15`. O valor equivale ao horário do último registro somado ao intervalo da sessão.
`SESSION_ROWTIME(<time-attr>, <gap-interval>)`	TIMESTAMP (atributo rowtime)	Retorna o horário de término da janela (exclusivo). Por exemplo, para a janela `(00:00, 00:15)`, o retorno é `00:14:59.999`. O valor retornado é um atributo rowtime que permite operações subsequentes baseadas em tempo. Aplica-se apenas a janelas de tempo de evento. Para mais detalhes, consulte Cascading windows.
`SESSION_PROCTIME(<time-attr>, <gap-interval>)`	TIMESTAMP (atributo de tempo de processamento)	Retorna o horário de término da janela (exclusivo). Por exemplo, para a janela `(00:00, 00:15)`, o retorno é `00:14:59.999`. O valor retornado é um atributo de tempo de processamento compatível com operações subsequentes baseadas em tempo. Aplica-se exclusivamente a janelas de tempo de processamento. Para mais detalhes, consulte Cascading windows.

Exemplo

Este exemplo conta cliques por usuário em cada sessão ativa, com um intervalo de sessão de 30 segundos e tempo de evento.

Dados de teste (tabela user_clicks)

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`

Instruções SQL

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;

Resultados

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