Realtime Compute for Apache Flink:Window aggregation

table.optimizer.window-rewrite-enabled: true

Syntaxe TVF de fenêtre SESSION : VVR 11.x vs VVR 8.x

La syntaxe TVF de fenêtre SESSION diffère selon les versions de VVR. Effectuez la mise à niveau vers VVR 11.1 ou ultérieure pour utiliser la syntaxe complète.

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;

Quand l'agrégation TVF de fenêtre revient à un mode sans TVF

Lorsqu'une requête inclut une TVF de fenêtre mais ne satisfait pas les conditions permettant de fusionner la TVF et l'agrégation, le système revient à un plan d'exécution sans TVF.

La TVF de fenêtre et l'instruction d'agrégation ne peuvent pas être fusionnées lorsque l'une des conditions suivantes est remplie :

1. Filtrage ou calcul sur les champs temporels de la fenêtre. window_start, window_end ou window_time est filtré ou modifié avant l'agrégation.

   -- 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. Une TVF de fenêtre est utilisée avec une fonction table définie par l'utilisateur (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. La clause GROUP BY omet 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. Une fonction d'agrégation définie par l'utilisateur Python (UDAF) est utilisée.

5. GROUPING SETS, CUBE ou ROLLUP effectuent un regroupement séparé par 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. Une fonction d'agrégation est appliquée à 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;

Condition 1 : La stratégie de phase d'agrégation autorise la double phase

table.optimizer.agg-phase-strategy est défini sur AUTO (valeur par défaut) ou TWO_PHASE. La définition sur ONE_PHASE désactive l'optimisation en deux phases.

Condition 2 : La fenêtre utilise le temps événementiel

La fenêtre doit utiliser le temps événementiel (rowtime). Les fenêtres en temps de traitement ne sont pas prises en charge.

Condition 3 : Le type de fenêtre n'est pas SESSION

Les types de fenêtre TUMBLE, HOP et CUMULATE sont pris en charge. Les fenêtres SESSION ne prennent pas en charge l'optimisation en deux phases.

Condition 4 : Toutes les fonctions d'agrégation prennent en charge la fusion partielle

Toutes les fonctions d'agrégation doivent prendre en charge l'opération de fusion. Les fonctions intégrées telles que SUM, COUNT, MIN, MAX et AVG sont prises en charge. Les UDAF personnalisées doivent implémenter la méthode merge().

Condition 5 : Le flux d'entrée est en insertion seule et la fenêtre peut être convertie au format TVF

Les conditions suivantes doivent toutes être remplies :

• Le flux d'entrée est en insertion seule.

• table.exec.emit.early-fire.enabled est défini sur false (valeur par défaut).

• table.exec.emit.late-fire.enabled est défini sur false (valeur par défaut).

• Pour les fenêtres HOP, la fenêtre doit être alignée (la taille de la fenêtre est divisible par l'intervalle de glissement).

Condition 6 : La distribution des données ne satisfait pas déjà aux exigences de partitionnement

La distribution des données d'entrée ne satisfait pas déjà aux exigences de partitionnement pour l'agrégation. Si les données sont déjà distribuées selon la clé de partitionnement, l'optimiseur détermine qu'aucune pré-agrégation supplémentaire n'est nécessaire et ne génère pas de nœud Local Aggregate.

Référence des paramètres