Realtime Compute for Apache Flink:Window aggregation

Parameter differences

VVR 8.x does not support the PARTITION BY clause. As a result, the SESSION window TVF uses the fields specified in the GROUP BY clause of the aggregation query to partition data, except the window_start, window_end, and window_time fields. In the following code, the sample query for Apache Flink 1.19 is equivalent to the sample query for VVR 8.x, which specifies that the SESSION window TVF uses the item field to partition data.

-- tables must have time attribute, e.g. `bidtime` in this table
> 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 |     |        |                                 |
+-------------+------------------------+------+-----+--------+---------------------------------+

-- Flink 1.19
> 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
> 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;

• Apache Flink 1.19 syntax

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

  Parameters:

  • data: a table that has time attribute columns.

  • keycols: column descriptors that specify which columns are used to partition the data in session windows.

  • timecol: the column descriptor that specifies which time attribute column is used for session windows.

  • gap: the maximum time interval between two events that belong to the same session window.

• VVR 8.x syntax

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

  Parameters:

  • data: a table that has time attribute columns.

  • timecol: the column descriptor that specifies which time attribute column is used for session windows.

  • gap: the maximum time interval between two events that belong to the same session window.

The SESSION window TVF in VVR 8.x must be used in an aggregation query. Otherwise, an error occurs. When the SESSION window TVF is used in an aggregation query, specific use cases are not supported. For more information, see Limits of window TVFs in aggregation queries.

Filtering or calculation of the window_start, window_end, or window_time field is performed. Example:

-- Filtering based 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;
  
-- Calculation of 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;

-- Calculation of 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;

A window TVF is used together with a user-defined table function (UDTF). Example:

> 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;

The GROUP BY clause of an aggregation query does not contain the window_start and window_end fields. Example:

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

A Python user-defined aggregate function (UDAF) is used.

GROUPING SETS, CUBE, or ROLLUP is specified for an aggregate function, which indicates that data is grouped separately by the window_start and window_end fields. Example:

> 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);

Aggregate functions are applied to the window_start, window_end, or window_time field. Example:

> 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;