Realtime Compute for Apache Flink:Visão geral das funções de janela

O Flink SQL suporta agregações de janela e agregações over. Este tópico aborda as agregações de janela. Essas agregações utilizam dois atributos de tempo — tempo de evento e tempo de processamento — e oferecem suporte às funções de janela TUMBLE, HOP e SESSION.

O Flink SQL suporta dois atributos de tempo: tempo de evento e tempo de processamento. O comportamento da janela varia conforme o atributo utilizado.

• Tempo de evento: geralmente corresponde ao timestamp incorporado em um registro.

  A janela se fecha quando a marca d'água ultrapassa o horário de término. A saída é gerada apenas quando os dados de fechamento chegam. Em uma única subtarefa, a marca d'água aumenta monotonicamente. Com múltiplas subtarefas ou tabelas de origem, o Flink utiliza o valor mínimo da marca d'água.

  Importante

  • Se existirem registros fora de ordem ou se uma subtarefa ou partição não tiver dados, a marca d'água não avança e a janela pode não fechar. Para resolver isso, especifique um deslocamento de marca d'água para dados fora de ordem e garanta que todas as subtarefas e partições recebam fluxo de dados. Caso uma partição esteja ociosa, adicione table.exec.source.idle-timeout: 10s ao campo Other Configuration em Parameters, na aba Configuration da página Deployments. Consulte Configuration para obter detalhes sobre os parâmetros.

  • Após o processamento dos dados com GROUP BY, operações JOIN entre dois fluxos de dados ou nós de janela OVER, a propriedade de marca d'água é perdida e o tempo de evento não pode mais ser usado para janelas.

• Tempo de processamento: horário do relógio do sistema no momento em que o Flink processa um evento.

  O Flink gera o tempo de processamento, que não existe nos dados brutos; portanto, defina explicitamente uma coluna de tempo de processamento.

  Nota

  Como o tempo de processamento depende da velocidade de chegada dos eventos e da ordem de processamento, os resultados de reprocessamento podem variar entre execuções.

O exemplo abaixo aplica uma janela tumbling de 1 hora sobre uma janela tumbling de 1 minuto:

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;