Use a TUMBLE function - Realtime Compute for Apache Flink: Central de Documentação do Alibaba Cloud

A função TUMBLE atribui cada linha de um fluxo de dados a uma janela fixa e sem sobreposição. Use-a em uma cláusula GROUP BY para calcular agregações por janela com base no tempo de evento ou no tempo de processamento.

Por exemplo, uma janela tumbling de 5 minutos aplicada a um fluxo de dados infinito gera intervalos discretos e consecutivos: [0:00, 0:05), [0:05, 0:10), [0:10, 0:15), e assim por diante.

Sintaxe

TUMBLE(<time-attr>, <size-interval>)
<size-interval>: INTERVAL 'string' timeUnit

O parâmetro <time-attr> deve ser um atributo de tempo válido: tempo de evento ou tempo de processamento. Para definir atributos de tempo, consulte Visão geral. Para a especificação completa, acesse a documentação do Apache Flink sobre Time Attributes.

Funções identificadoras de janela

Use estas funções na lista SELECT para obter os limites da janela ou propagar o atributo de tempo para janelas em cascata.

Função	Tipo de retorno	Descrição
`TUMBLE_START(time-attr, size-interval)`	TIMESTAMP	Início da janela (inclusivo). Para `[00:10, 00:15)`, retorna `00:10`.
`TUMBLE_END(time-attr, size-interval)`	TIMESTAMP	Fim da janela (inclusivo). Para `[00:00, 00:15]`, retorna `00:15`.
`TUMBLE_ROWTIME(time-attr, size-interval)`	TIMESTAMP (atributo rowtime)	Fim da janela (exclusivo). Para `(00:00, 00:15)`, retorna `00:14:59.999`. Retorna um atributo rowtime para uso como atributo de tempo em uma janela subsequente. Válido apenas para janelas de tempo de evento. Para janelas em cascata, consulte a Visão geral.
`TUMBLE_PROCTIME(time-attr, size-interval)`	TIMESTAMP (atributo rowtime)	Fim da janela (exclusivo). Para `(00:00, 00:15)`, retorna `00:14:59.999`. Retorna um atributo de tempo de processamento para uso em operações de tempo subsequentes. Válido apenas para janelas de tempo de processamento. Para janelas em cascata, consulte a Visão geral.

Exemplo 1: Contar cliques por usuário por minuto (tempo de evento)

Este exemplo conta quantas vezes cada usuário clica em uma URL dentro de cada janela tumbling de 1 minuto, usando tempo de evento com uma marca d'água de 2 segundos.

Dados de teste

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

Instruções de teste

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' = 'sls',
  ...
);

CREATE TEMPORARY TABLE tumble_output (
  window_start TIMESTAMP,
  window_end   TIMESTAMP,
  username     VARCHAR,
  clicks       BIGINT
) WITH (
  'connector' = 'datahub',
  ...
);

INSERT INTO tumble_output
SELECT
  TUMBLE_START(ts, INTERVAL '1' MINUTE) AS window_start,
  TUMBLE_END(ts, INTERVAL '1' MINUTE)   AS window_end,
  username,
  COUNT(click_url)
FROM user_clicks
GROUP BY TUMBLE(ts, INTERVAL '1' MINUTE), username;

Resultados do teste

window_start (TIMESTAMP)	window_end (TIMESTAMP)	username (VARCHAR)	clicks (BIGINT)
`2024-08-10 10:00:00.0`	`2024-08-10 10:01:00.0`	Jark	3
`2024-08-10 10:01:00.0`	`2024-08-10 10:02:00.0`	Jark	2
`2024-08-10 10:02:00.0`	`2024-08-10 10:03:00.0`	Timo	1

Exemplo 2: Contar cliques por usuário por minuto (tempo de processamento)

Este exemplo usa tempo de processamento em vez de tempo de evento. A função PROCTIME() gera automaticamente a coluna de tempo de processamento, eliminando a necessidade de marca d'água.

Nota

O Simple Log Service (SLS) aceita apenas colunas do tipo VARCHAR em instruções DDL; portanto, este exemplo grava a saída no DataHub.

Dados de teste

username (VARCHAR)	click_url (VARCHAR)
Jark	`http://taobao.com/xxx`
Jark	`http://taobao.com/xxx`
Jark	`http://taobao.com/xxx`
Jark	`http://taobao.com/xxx`
Jark	`http://taobao.com/xxx`
Timo	`http://taobao.com/xxx`

Instruções de teste

CREATE TEMPORARY TABLE window_test (
  username  VARCHAR,
  click_url VARCHAR,
  ts        AS PROCTIME()
) WITH (
  'connector' = 'sls',
  ...
);

CREATE TEMPORARY TABLE tumble_output (
  window_start TIMESTAMP,
  window_end   TIMESTAMP,
  username     VARCHAR,
  clicks       BIGINT
) WITH (
  'connector' = 'datahub',
  ...
);

INSERT INTO tumble_output
SELECT
  TUMBLE_START(ts, INTERVAL '1' MINUTE),
  TUMBLE_END(ts, INTERVAL '1' MINUTE),
  username,
  COUNT(click_url)
FROM window_test
GROUP BY TUMBLE(ts, INTERVAL '1' MINUTE), username;

Resultados do teste

window_start (TIMESTAMP)	window_end (TIMESTAMP)	username (VARCHAR)	clicks (BIGINT)
`2024-08-10 14:43:00.000`	`2024-08-10 14:44:00.000`	Jark	5
`2024-08-10 14:43:00.000`	`2024-08-10 14:44:00.000`	Timo	1

Próximos passos

Visão geral — saiba mais sobre atributos de tempo e janelas em cascata