Flink SQL 知其所以然：Window TopN 操作

作者：antigeneral了呀 2022-07-05 09:03:05

今天我們來學習 Flink SQL 中的 Window TopN 操作。

大家好，我是老羊，今天我們來學習 Flink SQL 中的 Window TopN 操作。

Window TopN 定義（支持 Streaming）：Window TopN 是一種特殊的 TopN，它的返回結果是每一個窗口內的 N 個最小值或者最大值。
應用場景：小伙伴萌會問了，我有了 TopN 為啥還需要 Window TopN 呢？還記得上文介紹 TopN 說道的 TopN 時會出現中間結果，從而出現回撤數據的嘛？Window TopN 不會出現回撤數據，因為 Window TopN 實現是在窗口結束時輸出最終結果，不會產生中間結果。而且注意，因為是窗口上面的操作，Window TopN 在窗口結束時，會自動把 State 給清除。
SQL 語法標準：

SELECT [column_list]
FROM (
   SELECT [column_list],
     ROW_NUMBER() OVER (PARTITION BY window_start, window_end [, col_key1...]
       ORDER BY col1 [asc|desc][, col2 [asc|desc]...]) AS rownum
   FROM table_name) -- windowing TVF
WHERE rownum <= N [AND conditions]

實際案例：取當前這一分鐘的搜索關鍵詞下的搜索熱度前 10 名的詞條數據。

輸入表字段：

-- 字段名         備注
-- key              搜索關鍵詞
-- name             搜索熱度名稱
-- search_cnt       熱搜消費熱度（比如 3000）
-- timestamp        消費詞條時間戳
CREATE TABLE source_table (
    name BIGINT NOT NULL,
    search_cnt BIGINT NOT NULL,
    key BIGINT NOT NULL,
    row_time AS cast(CURRENT_TIMESTAMP as timestamp(3)),
    WATERMARK FOR row_time AS row_time
) WITH (
  ...
);
-- 輸出表字段：
-- 字段名         備注
-- key              搜索關鍵詞
-- name             搜索熱度名稱
-- search_cnt       熱搜消費熱度（比如 3000）
-- window_start     窗口開始時間戳
-- window_end       窗口結束時間戳
CREATE TABLE sink_table (
    key BIGINT,
    name BIGINT,
    search_cnt BIGINT,
    window_start TIMESTAMP(3),
    window_end TIMESTAMP(3)
) WITH (
  ...
);
-- 處理 sql：
INSERT INTO sink_table
SELECT key, name, search_cnt, window_start, window_end
FROM (
   SELECT key, name, search_cnt, window_start, window_end, 
     ROW_NUMBER() OVER (PARTITION BY window_start, window_end, key
       ORDER BY search_cnt desc) AS rownum
   FROM (
      SELECT window_start, window_end, key, name, max(search_cnt) as search_cnt
      -- window tvf 寫法
      FROM TABLE(TUMBLE(TABLE source_table, DESCRIPTOR(row_time), INTERVAL '1' MINUTES))
      GROUP BY window_start, window_end, key, name
   )
)
WHERE rownum <= 100

輸出結果：

+I[關鍵詞1, 詞條1, 8670, 2021-1-28T22:34, 2021-1-28T22:35]
+I[關鍵詞1, 詞條2, 6928, 2021-1-28T22:34, 2021-1-28T22:35]
+I[關鍵詞1, 詞條3, 1735, 2021-1-28T22:34, 2021-1-28T22:35]
+I[關鍵詞1, 詞條4, 7287, 2021-1-28T22:34, 2021-1-28T22:35]
...

可以看到結果是符合預期的，其中沒有回撤數據。

SQL 語義。

數據源：數據源即最新的詞條下面的搜索詞的搜索熱度數據，消費到 Kafka 中數據后，將數據按照窗口聚合的 key 通過 hash 分發策略發送到下游窗口聚合算子。
窗口聚合算子：進行窗口聚合計算，隨著時間的推進，將窗口聚合結果計算完成發往下游窗口排序算子。
窗口排序算子：這個算子其實也是一個窗口算子，只不過這個窗口算子為每個 Key 維護了一個 TopN 的榜單數據，接受到上游發送的窗口結果數據進行排序，隨著時間的推進，窗口的結束，將排序的結果輸出到下游數據匯算子。
數據匯：接收到上游的數據之后，然后輸出到外部存儲引擎中。

責任編輯：姜華來源：大數據羊說

Flink SQL TopN