Window Top-N

Batch Streaming

Window Top-N is a special Top-N which returns the N smallest or largest values for each window and other partitioned keys.

For streaming queries, unlike regular Top-N on continuous tables, window Top-N does not emit intermediate results but only a final result, the total top N records at the end of the window. Moreover, window Top-N purges all intermediate state when no longer needed. Therefore, window Top-N queries have better performance if users don’t need results updated per record. Usually, Window Top-N is used with Windowing TVF directly. Besides, Window Top-N could be used with other operations based on Windowing TVF, such as Window Aggregation, Window TopN and Window Join.

Window Top-N can be defined in the same syntax as regular Top-N, see Top-N documentation for more information. Besides that, Window Top-N requires the PARTITION BY clause contains window_start and window_end columns of the relation applied Windowing TVF or Window Aggregation. Otherwise, the optimizer won’t be able to translate the query.

The following shows the syntax of the Window Top-N statement:

  1. SELECT [column_list]
  2. FROM (
  3.    SELECT [column_list],
  4.      ROW_NUMBER() OVER (PARTITION BY window_start, window_end [, col_key1...]
  5.        ORDER BY col1 [asc|desc][, col2 [asc|desc]...]) AS rownum
  6.    FROM table_name) -- relation applied windowing TVF
  7. WHERE rownum <= N [AND conditions]

Example

Window Top-N follows after Window Aggregation

The following example shows how to calculate Top 3 suppliers who have the highest sales for every tumbling 10 minutes window.

  1. -- tables must have time attribute, e.g. `bidtime` in this table
  2. Flink SQL> desc Bid;
  3. +-------------+------------------------+------+-----+--------+---------------------------------+
  4. |        name |                   type | null | key | extras |                       watermark |
  5. +-------------+------------------------+------+-----+--------+---------------------------------+
  6. |     bidtime | TIMESTAMP(3) *ROWTIME* | true |     |        | `bidtime` - INTERVAL '1' SECOND |
  7. |       price |         DECIMAL(10, 2) | true |     |        |                                 |
  8. |        item |                 STRING | true |     |        |                                 |
  9. | supplier_id |                 STRING | true |     |        |                                 |
  10. +-------------+------------------------+------+-----+--------+---------------------------------+
  12. Flink SQL> SELECT * FROM Bid;
  13. +------------------+-------+------+-------------+
  14. |          bidtime | price | item | supplier_id |
  15. +------------------+-------+------+-------------+
  16. | 2020-04-15 08:05 |  4.00 |    A |   supplier1 |
  17. | 2020-04-15 08:06 |  4.00 |    C |   supplier2 |
  18. | 2020-04-15 08:07 |  2.00 |    G |   supplier1 |
  19. | 2020-04-15 08:08 |  2.00 |    B |   supplier3 |
  20. | 2020-04-15 08:09 |  5.00 |    D |   supplier4 |
  21. | 2020-04-15 08:11 |  2.00 |    B |   supplier3 |
  22. | 2020-04-15 08:13 |  1.00 |    E |   supplier1 |
  23. | 2020-04-15 08:15 |  3.00 |    H |   supplier2 |
  24. | 2020-04-15 08:17 |  6.00 |    F |   supplier5 |
  25. +------------------+-------+------+-------------+
  27. Flink SQL> SELECT *
  28.   FROM (
  29.     SELECT *, ROW_NUMBER() OVER (PARTITION BY window_start, window_end ORDER BY price DESC) as rownum
  30.     FROM (
  31.       SELECT window_start, window_end, supplier_id, SUM(price) as price, COUNT(*) as cnt
  32.       FROM TABLE(
  33.         TUMBLE(TABLE Bid, DESCRIPTOR(bidtime), INTERVAL '10' MINUTES))
  34.       GROUP BY window_start, window_end, supplier_id
  35.     )
  36.   ) WHERE rownum <= 3;
  37. +------------------+------------------+-------------+-------+-----+--------+
  38. |     window_start |       window_end | supplier_id | price | cnt | rownum |
  39. +------------------+------------------+-------------+-------+-----+--------+
  40. | 2020-04-15 08:00 | 2020-04-15 08:10 |   supplier1 |  6.00 |   2 |      1 |
  41. | 2020-04-15 08:00 | 2020-04-15 08:10 |   supplier4 |  5.00 |   1 |      2 |
  42. | 2020-04-15 08:00 | 2020-04-15 08:10 |   supplier2 |  4.00 |   1 |      3 |
  43. | 2020-04-15 08:10 | 2020-04-15 08:20 |   supplier5 |  6.00 |   1 |      1 |
  44. | 2020-04-15 08:10 | 2020-04-15 08:20 |   supplier2 |  3.00 |   1 |      2 |
  45. | 2020-04-15 08:10 | 2020-04-15 08:20 |   supplier3 |  2.00 |   1 |      3 |
  46. +------------------+------------------+-------------+-------+-----+--------+

Note: in order to better understand the behavior of windowing, we simplify the displaying of timestamp values to not show the trailing zeros, e.g. 2020-04-15 08:05 should be displayed as 2020-04-15 08:05:00.000 in Flink SQL Client if the type is TIMESTAMP(3).

Window Top-N follows after Windowing TVF

The following example shows how to calculate Top 3 items which have the highest price for every tumbling 10 minutes window.

  1. Flink SQL> SELECT *
  2.   FROM (
  3.     SELECT bidtime, price, item, supplier_id, window_start, window_end, ROW_NUMBER() OVER (PARTITION BY window_start, window_end ORDER BY price DESC) as rownum
  4.     FROM TABLE(
  5.                TUMBLE(TABLE Bid, DESCRIPTOR(bidtime), INTERVAL '10' MINUTES))
  6.   ) WHERE rownum <= 3;
  7. +------------------+-------+------+-------------+------------------+------------------+--------+
  8. |          bidtime | price | item | supplier_id |     window_start |       window_end | rownum |
  9. +------------------+-------+------+-------------+------------------+------------------+--------+
  10. | 2020-04-15 08:05 |  4.00 |    A |   supplier1 | 2020-04-15 08:00 | 2020-04-15 08:10 |      2 |
  11. | 2020-04-15 08:06 |  4.00 |    C |   supplier2 | 2020-04-15 08:00 | 2020-04-15 08:10 |      3 |
  12. | 2020-04-15 08:09 |  5.00 |    D |   supplier4 | 2020-04-15 08:00 | 2020-04-15 08:10 |      1 |
  13. | 2020-04-15 08:11 |  2.00 |    B |   supplier3 | 2020-04-15 08:10 | 2020-04-15 08:20 |      3 |
  14. | 2020-04-15 08:15 |  3.00 |    H |   supplier2 | 2020-04-15 08:10 | 2020-04-15 08:20 |      2 |
  15. | 2020-04-15 08:17 |  6.00 |    F |   supplier5 | 2020-04-15 08:10 | 2020-04-15 08:20 |      1 |
  16. +------------------+-------+------+-------------+------------------+------------------+--------+

Note: in order to better understand the behavior of windowing, we simplify the displaying of timestamp values to not show the trailing zeros, e.g. 2020-04-15 08:05 should be displayed as 2020-04-15 08:05:00.000 in Flink SQL Client if the type is TIMESTAMP(3).

Limitation

Currently, Flink only supports Window Top-N follows after Windowing TVF with Tumble Windows, Hop Windows and Cumulate Windows. Window Top-N follows after Windowing TVF with Session windows will be supported in the near future.