Realtime Compute for Apache Flink:Window aggregation

table.optimizer.window-rewrite-enabled: true

Sintaxis de TVF de ventanas SESSION: VVR 11.x frente a VVR 8.x

La sintaxis de TVF de ventanas SESSION varía entre versiones de VVR. Actualice a VVR 11.1 o posterior para utilizar la sintaxis con todas las funciones disponibles.

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;

Cuándo la agregación TVF de ventanas vuelve al modo sin TVF

Cuando una consulta incluye una TVF de ventanas pero no cumple las condiciones para combinar la TVF con la agregación, el sistema vuelve a un plan de ejecución sin TVF.

La TVF de ventanas y la instrucción de agregación no se pueden combinar cuando se cumple alguna de las siguientes condiciones:

1. Filtrado o cálculo sobre campos de tiempo de ventana. Se filtra o modifica window_start, window_end o window_time antes de la agregación.

   -- 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. Se utiliza una TVF de ventanas junto con una función con valores de tabla definida por el usuario (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 la cláusula GROUP BY le falta window_start o 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. Se utiliza una función de agregación definida por el usuario (UDAF) de Python.

5. GROUPING SETS, CUBE o ROLLUP agrupan por separado por window_start o 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. Se aplica una función de agregación a window_start, window_end o 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;

Condición 1: La estrategia de fase de agregación permite dos fases

table.optimizer.agg-phase-strategy está establecido en AUTO (predeterminado) o TWO_PHASE. Establecerlo en ONE_PHASE deshabilita la optimización de dos fases.

Condición 2: La ventana utiliza tiempo de evento

La ventana debe utilizar tiempo de evento (rowtime). Las ventanas de tiempo de procesamiento no son compatibles.

Condición 3: El tipo de ventana no es SESSION

Los tipos de ventana TUMBLE, HOP y CUMULATE son compatibles. Las ventanas SESSION no admiten la optimización de dos fases.

Condición 4: Todas las funciones de agregación admiten combinación parcial

Todas las funciones de agregación deben admitir la operación de combinación. Las funciones integradas como SUM, COUNT, MIN, MAX y AVG son compatibles. Las UDAF personalizadas deben implementar el método merge().

Condición 5: El flujo de entrada es de solo inserción y la ventana se puede convertir a formato TVF

Deben cumplirse todas las condiciones siguientes:

• El flujo de entrada es de solo inserción.

• table.exec.emit.early-fire.enabled está establecido en false (predeterminado).

• table.exec.emit.late-fire.enabled está establecido en false (predeterminado).

• Para ventanas HOP, la ventana debe estar alineada (el tamaño de la ventana debe ser divisible por el intervalo de deslizamiento).

Condición 6: La distribución de datos no satisface ya los requisitos de particionamiento

La distribución de datos de entrada no cumple todavía con los requisitos de particionamiento para la agregación. Si los datos ya están distribuidos por la clave de partición, el optimizador determina que no se necesita preagregación adicional y no genera un nodo Local Aggregate.

Referencia de parámetros