Flink SQL Window TopN 详解

Window TopN 定义（⽀持 Streaming）： Window TopN 是特殊的 TopN，返回结果是每⼀个窗⼝内的 N 个最⼩值或者最⼤值。

应⽤场景： 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]
1
2
3
4
5
6
7

实际案例： 取当前这⼀分钟的搜索关键词下的搜索热度前 10 名的词条数据。

输⼊表字段：

-- 字段名 备注
-- key 搜索关键词
-- name 搜索热度名称
-- search_cnt 热搜消费热度（⽐如 3000）
-- timestamp 消费词条时间戳
CREATE TABLE source_table (
 name STRING NOT NULL,
 search_cnt BIGINT NOT NULL,
 key STRING NOT NULL,
 row_time timestamp(3),
 WATERMARK FOR row_time AS row_time
) WITH (
 'connector' = 'filesystem', 
 'path' = 'file:///Users/hhx/Desktop/source_table.csv',
 'format' = 'csv'
);

A,100,a,2021-11-01 00:01:00
B,200,b,2021-11-01 00:01:00
C,300,c,2021-11-01 00:01:00
D,400,d,2021-11-01 00:01:00
A,200,a,2021-11-01 00:01:05
B,300,b,2021-11-01 00:01:05
C,400,c,2021-11-01 00:01:05
D,500,d,2021-11-01 00:01:05
A,300,a,2021-11-01 00:02:00
B,400,b,2021-11-01 00:02:00
C,500,c,2021-11-01 00:02:00
D,600,d,2021-11-01 00:02:00

-- 输出表字段：
-- 字段名 备注
-- 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 (
 ...
);

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' MINUTE))
 GROUP BY window_start, window_end, key, name
 )
)
WHERE rownum <= 2;
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64

输出结果：

SQL 转换为算子：

• 数据源 ：数据源即最新的词条下⾯的搜索词的搜索热度数据，消费到 Kafka 后，将数据按照窗⼝聚合的 key 通过 hash 分发策略发送到下游窗⼝聚合算⼦。
• 窗⼝聚合算⼦ ：进⾏窗⼝聚合计算，随着时间的推进，将窗⼝聚合结果计算完成发往下游窗⼝排序算⼦。
• 窗⼝排序算⼦ ：这个算⼦其实也是⼀个窗⼝算⼦，只不过这个窗⼝算⼦为每个 Key 维护了⼀个 TopN 的榜单数据，接受到上游发送的窗⼝结果数据进⾏排序，随着时间的推进，窗⼝的结束，将排序的结果输出到下游数据汇算⼦。
• 数据汇 ：接收到上游的数据之后，然后输出到外部存储引擎中。