Flink SQL 知其所以然：TopN、Order By、Limit 操作
DML：Order By、Limit 子句
大家好，我是老羊，今天我們來學習 Flink SQL 中的 TopN、Order By、Limit 3個操作。
1.Order By 子句
支持 Batch\Streaming，但在實時任務中一般用的非常少。
實時任務中，Order By 子句中必須要有時間屬性字段，并且時間屬性必須為升序時間屬性，即 WATERMARK FOR rowtime_column AS rowtime_column - INTERVAL '0.001' SECOND 或者 WATERMARK FOR rowtime_column AS rowtime_column。
舉例：
CREATE TABLE source_table_1 (
user_id BIGINT NOT NULL,
row_time AS cast(CURRENT_TIMESTAMP as timestamp(3)),
WATERMARK FOR row_time AS row_time
) WITH (
'connector' = 'datagen',
'rows-per-second' = '10',
'fields.user_id.min' = '1',
'fields.user_id.max' = '10'
);
CREATE TABLE sink_table (
user_id BIGINT
) WITH (
'connector' = 'print'
);
INSERT INTO sink_table
SELECT user_id
FROM source_table_1
Order By row_time, user_id desc
2.Limit 子句
支持 Batch\Streaming，但實時場景一般不使用，但是此處依然舉一個例子：
CREATE TABLE source_table_1 (
user_id BIGINT NOT NULL,
row_time AS cast(CURRENT_TIMESTAMP as timestamp(3)),
WATERMARK FOR row_time AS row_time
) WITH (
'connector' = 'datagen',
'rows-per-second' = '10',
'fields.user_id.min' = '1',
'fields.user_id.max' = '10'
);
CREATE TABLE sink_table (
user_id BIGINT
) WITH (
'connector' = 'print'
);
INSERT INTO sink_table
SELECT user_id
FROM source_table_1
Limit 3
結果如下，只有 3 條輸出：
+I[5]
+I[9]
+I[4]
DML：TopN 子句
- TopN 定義（支持 Batch\Streaming）：TopN 其實就是對應到離線數倉中的 row_number()，可以使用 row_number() 對某一個分組的數據進行排序
- 應用場景：根據 某個排序 條件，計算某個分組下的排行榜數據
- SQL 語法標準：
SELECT [column_list]
FROM (
SELECT [column_list],
ROW_NUMBER() OVER ([PARTITION BY col1[, col2...]]
ORDER BY col1 [asc|desc][, col2 [asc|desc]...]) AS rownum
FROM table_name)
WHERE rownum <= N [AND conditions]
ROW_NUMBER()：標識 TopN 排序子句
PARTITION BY col1[, col2...]：標識分區字段，代表按照這個 col 字段作為分區粒度對數據進行排序取 topN，比如下述案例中的partition by key，就是根據需求中的搜索關鍵詞（key）做為分區
ORDER BY col1 [asc|desc][, col2 [asc|desc]...]：標識 TopN 的排序規則，是按照哪些字段、順序或逆序進行排序
WHERE rownum <= N：這個子句是一定需要的，只有加上了這個子句，Flink 才能將其識別為一個 TopN 的查詢，其中 N 代表 TopN 的條目數
[AND conditions]：其他的限制條件也可以加上
- 實際案例：取某個搜索關鍵詞下的搜索熱度前 10 名的詞條數據。
輸入數據為搜索詞條數據的搜索熱度數據，當搜索熱度發生變化時，會將變化后的數據寫入到數據源的 Kafka 中：
數據源 schema：
-- 字段名 備注
-- 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 (
...
);
-- 數據匯 schema：
-- key 搜索關鍵詞
-- name 搜索熱度名稱
-- search_cnt 熱搜消費熱度（比如 3000）
-- timestamp 消費詞條時間戳
CREATE TABLE sink_table (
key BIGINT,
name BIGINT,
search_cnt BIGINT,
`timestamp` TIMESTAMP(3)
) WITH (
...
);
-- DML 邏輯
INSERT INTO sink_table
SELECT key, name, search_cnt, row_time as `timestamp`
FROM (
SELECT key, name, search_cnt, row_time,
-- 根據熱搜關鍵詞 key 作為 partition key，然后按照 search_cnt 倒排取前 100 名
ROW_NUMBER() OVER (PARTITION BY key
ORDER BY search_cnt desc) AS rownum
FROM source_table)
WHERE rownum <= 100
輸出結果：
-D[關鍵詞1, 詞條1, 4944]
+I[關鍵詞1, 詞條1, 8670]
+I[關鍵詞1, 詞條2, 1735]
-D[關鍵詞1, 詞條3, 6641]
+I[關鍵詞1, 詞條3, 6928]
-D[關鍵詞1, 詞條4, 6312]
+I[關鍵詞1, 詞條4, 7287]
可以看到輸出數據是有回撤數據的，為什么會出現回撤，我們來看看 SQL 語義。
- SQL 語義
上面的 SQL 會翻譯成以下三個算子：
數據源：數據源即最新的詞條下面的搜索詞的搜索熱度數據，消費到 Kafka 中數據后，按照 partition key 將數據進行 hash 分發到下游排序算子，相同的 key 數據將會發送到一個并發中
排序算子：為每個 Key 維護了一個 TopN 的榜單數據，接受到上游的一條數據后，如果 TopN 榜單還沒有到達 N 條，則將這條數據加入 TopN 榜單后，直接下發數據，如果到達 N 條之后，經過 TopN 計算，發現這條數據比原有的數據排序靠前，那么新的 TopN 排名就會有變化，就變化了的這部分數據之前下發的排名數據撤回（即回撤數據），然后下發新的排名數據
數據匯：接收到上游的數據之后，然后輸出到外部存儲引擎中
上面三個算子也是會 24 小時一直運行的。
