Realtime Compute for Apache Flink:Agregação de janela

table.optimizer.window-rewrite-enabled: true

Sintaxe da TVF de janela SESSION: VVR 11.x vs VVR 8.x

A sintaxe da TVF de janela SESSION varia entre as versões do VVR. Atualize para o VVR 11.1 ou posterior para utilizar a sintaxe completa.

SESSION(TABLE data [PARTITION BY(keycols, ...)], DESCRIPTOR(timecol), gap)

SESSION(TABLE data, DESCRIPTOR(timecol), gap)

-- The Bid table schema (used in all examples below)
> desc Bid;
+-------------+------------------------+------+-----+--------+---------------------------------+
|        name |                   type | null | key | extras |                       watermark |
+-------------+------------------------+------+-----+--------+---------------------------------+
|     bidtime | TIMESTAMP(3) *ROWTIME* | true |     |        | `bidtime` - INTERVAL '1' SECOND |
|       price |         DECIMAL(10, 2) | true |     |        |                                 |
|        item |                 STRING | true |     |        |                                 |
+-------------+------------------------+------+-----+--------+---------------------------------+

-- VVR 11.x: partition field declared explicitly in SESSION()
> SELECT window_start, window_end, item, SUM(price) AS total_price
  FROM TABLE(
      SESSION(TABLE Bid PARTITION BY item, DESCRIPTOR(bidtime), INTERVAL '5' MINUTES))
  GROUP BY item, window_start, window_end;

-- VVR 8.x: partition field inferred from GROUP BY
> SELECT window_start, window_end, item, SUM(price) AS total_price
  FROM TABLE(
      SESSION(TABLE Bid, DESCRIPTOR(bidtime), INTERVAL '5' MINUTES))
  GROUP BY item, window_start, window_end;

Quando a agregação por TVF de janela reverte para o modo não-TVF

Quando uma consulta inclui uma TVF de janela, mas não atende às condições para mesclar a TVF e a agregação, o sistema reverte para um plano de execução não-TVF.

A TVF de janela e a instrução de agregação não podem ser mescladas quando qualquer uma das seguintes condições for atendida:

1. Filtragem ou computação em campos de tempo da janela. window_start, window_end ou window_time é filtrado ou modificado antes da agregação.

   -- Filtering on window_start
   > SELECT window_start, window_end, item, SUM(price) AS total_price
       FROM
       (SELECT item, price, window_start, window_end FROM
       TABLE(
       SESSION(TABLE Bid, DESCRIPTOR(bidtime), INTERVAL '5' MINUTES))
       WHERE window_start >= TIMESTAMP '2020-04-15 08:06:00.000')
       GROUP BY item, window_start, window_end;
   
   -- Arithmetic on window_start
   > SELECT window_start, window_end, item, SUM(price) AS total_price
       FROM
       (SELECT item, price, window_start + (INTERVAL '1' SECOND) AS window_start, window_end FROM
       TABLE(
       SESSION(TABLE Bid, DESCRIPTOR(bidtime), INTERVAL '5' MINUTES)))
       GROUP BY item, window_start, window_end;
   
   -- Type casting on window_start
   > SELECT window_start, window_end, item, SUM(price) AS total_price
       FROM
       (SELECT item, price, CAST(window_start AS varchar) AS window_start, window_end FROM
       TABLE(
       SESSION(TABLE Bid, DESCRIPTOR(bidtime), INTERVAL '5' MINUTES)))
       GROUP BY item, window_start, window_end;

2. Uma TVF de janela é usada com uma função com valor de tabela definida pelo usuário (UDTF).

   > SELECT window_start, window_end, category, SUM(price) AS total_price
       FROM
       (SELECT category, price, window_start, window_end FROM
       TABLE(
       SESSION(TABLE Bid, DESCRIPTOR(bidtime), INTERVAL '5' MINUTES)),
       LATERAL TABLE(category_udtf(item)) AS T(category))
       GROUP BY category, window_start, window_end;

3. A cláusula GROUP BY não contém window_start ou window_end.

   > SELECT window_start, item, SUM(price) AS total_price
     FROM TABLE(
         SESSION(TABLE Bid, DESCRIPTOR(bidtime), INTERVAL '5' MINUTES))
     GROUP BY item, window_start;

4. Uma função de agregação definida pelo usuário (UDAF) Python é utilizada.

5. GROUPING SETS, CUBE ou ROLLUP agrupam separadamente por window_start ou window_end.

   > SELECT item, SUM(price) AS total_price
     FROM TABLE(
         SESSION(TABLE Bid, DESCRIPTOR(bidtime), INTERVAL '5' MINUTES))
     GROUP BY GROUPING SETS((item), (window_start), (window_end));
   
   > SELECT item, SUM(price) AS total_price
     FROM TABLE(
         SESSION(TABLE Bid, DESCRIPTOR(bidtime), INTERVAL '5' MINUTES))
     GROUP BY CUBE (item, window_start, window_end);
   
   > SELECT item, SUM(price) AS total_price
     FROM TABLE(
         SESSION(TABLE Bid, DESCRIPTOR(bidtime), INTERVAL '5' MINUTES))
     GROUP BY ROLLUP (item, window_start, window_end);

6. Uma função de agregação é aplicada a window_start, window_end ou window_time.

   > SELECT window_start, window_end, item, SUM(price) AS total_price, MAX(window_end) AS max_end
     FROM TABLE(
         SESSION(TABLE Bid, DESCRIPTOR(bidtime), INTERVAL '5' MINUTES))
     GROUP BY item, window_start, window_end;

Condição 1: A estratégia de fase de agregação permite duas fases

table.optimizer.agg-phase-strategy está definido como AUTO (padrão) ou TWO_PHASE. Defina como ONE_PHASE desabilita a otimização de duas fases.

Condição 2: A janela usa event time

A janela deve usar event time (rowtime). Janelas de processing-time não são suportadas.

Condição 3: O tipo de janela não é SESSION

Os tipos de janela TUMBLE, HOP e CUMULATE são suportados. Janelas SESSION não oferecem suporte à otimização de duas fases.

Condição 4: Todas as funções de agregação suportam mesclagem parcial

Todas as funções de agregação devem suportar a operação de mesclagem. Funções integradas como SUM, COUNT, MIN, MAX e AVG são suportadas. UDAFs personalizadas devem implementar o método merge().

Condição 5: A stream de entrada é insert-only e a janela pode ser convertida para a forma TVF

As seguintes condições devem ser atendidas simultaneamente:

• A stream de entrada é insert-only.

• table.exec.emit.early-fire.enabled está definido como false (padrão).

• table.exec.emit.late-fire.enabled está definido como false (padrão).

• Para janelas HOP, a janela deve estar alinhada (o tamanho da janela deve ser divisível pelo intervalo de deslizamento).

Condição 6: A distribuição de dados ainda não satisfaz os requisitos de particionamento

A distribuição dos dados de entrada ainda não atende aos requisitos de particionamento para agregação. Se os dados já estiverem distribuídos pela chave de partição, o otimizador determina que nenhuma pré-agregação adicional é necessária e não gera um nó Local Aggregate.

Referência de parâmetros