Flink SQL知其所以然：大家都用 Cumulate Window 計算累計指標啦
1.序篇
此節就是窗口聚合章節的第三篇，上節介紹了 1.13 window tvf tumble window 實現，本節主要介紹 1.13. window tvf 的一個重磅更新，即 cumulate window。
本節從以下幾個章節給大家詳細介紹 cumulate window 的能力。
- 應用場景介紹
- 預期的效果
- 解決方案介紹
- 總結及展望篇
2.應用場景介紹
先來一個簡單的小調查：在實時場景中，你見到過最多的指標需求場景是哪一種?
答案：博主相信，占比比較多的不是 PCU(即同時在線 PV，UV)，而是周期內累計 PV，UV 指標(如每天累計到當前這一分鐘的 PV，UV)。因為這類指標是一段周期內的累計狀態，對分析師來說更具統計分析價值，而且幾乎所有的復合指標都是基于此類指標的統計(不然離線為啥都要一天的數據，而不要一分鐘的數據呢)。
本文要介紹的就是周期內累計 PV，UV 指標在 flink 1.13 版本的最優解決方案。
3.預期的效果
先來一個實際案例來看看在具體輸入值的場景下，輸出值應該長啥樣。
指標：每天的截止當前分鐘的累計 money(sum(money))，去重 id 數(count(distinct id))。每天代表窗口大小為 1 天，分鐘代表移動步長為分鐘級別。
來一波輸入數據：
預期輸出數據：
轉化為折線圖長這樣：
當日累計
可以看到，其特點就在于，每一分鐘的輸出結果都是當天零點累計到當前的結果。
4.解決方案介紹
4.1.flink 1.13 之前
可選的解決方案有兩種
- tumble window(1天窗口) + early-fire(1分鐘)
- group by(1天) + minibatch(1分鐘)
但是上述兩種解決方案產出的都是 retract 流，關于 retract 流存在的缺點見如下文章：
踩坑記 | flink sql count 還有這種坑!
并且 tumble window + early-fire 的觸發機制是基于處理時間而非事件時間，具體缺點見如下文章：
https://mp.weixin.qq.com/s/L8-RSS6v3Ppts60CWngiOA
4.2.flink 1.13 及之后
誕生了 cumulate window 解法，具體見官網鏈接：
https://nightlies.apache.org/flink/flink-docs-release-1.13/docs/dev/table/sql/queries/window-tvf/#cumulate
如下官網文檔所示，介紹 cumulate window 的第一句話就是 cumulate window 非常適合于之前使用 tumble window + early-fire 的場景?？梢哉f cumulate window 就是在用戶計算周期內累計 PV，UV 指標時，使用了 tumble window + early-fire 后發現這種方案存在了很多坑的情況下，而誕生的!
cumulate window
其計算機制如下圖所示：
cumulate window
還是以剛剛的案例說明，以天為窗口，每分鐘輸出一次當天零點到當前分鐘的累計值，在 cumulate window 中，其窗口劃分規則如下：
- [2021-11-01 00:00:00, 2021-11-01 00:01:00]
- [2021-11-01 00:00:00, 2021-11-01 00:02:00]
- [2021-11-01 00:00:00, 2021-11-01 00:03:00] ...
- [2021-11-01 00:00:00, 2021-11-01 23:58:00]
- [2021-11-01 00:00:00, 2021-11-01 23:59:00]
第一個 window 統計的是一個區間的數據;第二個 window 統計的是第一區間和第二個區間的數據;第三個 window 統計的是第一區間，第二個區間和第三個區間的數據。
那么以 cumulate window 實現上述的需求，具體的 SQL 如下：
- SELECT UNIX_TIMESTAMP(CAST(window_end AS STRING)) * 1000 as window_end,
- window_start,
- sum(money) as sum_money,
- count(distinct id) as count_distinct_id
- FROM TABLE(CUMULATE(
- TABLE source_table
- , DESCRIPTOR(row_time)
- , INTERVAL '60' SECOND
- , INTERVAL '1' DAY))
- GROUP BY window_start,
- window_end
其中 CUMULATE(TABLE source_table, DESCRIPTOR(row_time), INTERVAL '60' SECOND, INTERVAL '1' DAY) 中的INTERVAL '1' DAY 代表窗口大小為 1 天，INTERVAL '60' SECOND，窗口劃分步長為 60s。
其中 window_start, window_end 字段是 cumulate window 自動生成的類型是 timestamp(3)。
window_start 固定為窗口的開始時間。window_end 為一個子窗口的結束時間。
最終結果如下。
輸入數據：
輸出數據：
Notes：天級別窗口劃分的時候一定要注意時區問題喔!https://nightlies.apache.org/flink/flink-docs-master/zh/docs/dev/table/timezone/
4.3.cumulate window 原理解析
首先 cumulate window 是一個窗口，其窗口計算的觸發也是完全由 watermark 推動的。與 tumble window 一樣。
以上述天窗口分鐘累計案例舉例：cumulate window 維護了一個 slice state 和 merged state，slice state 就是每一分鐘內窗口數據(叫做切片)，merged state 的作用是當 watermark 推動到下一分鐘時，這一分鐘的 slice state 就會被 merge 到 merged stated 中，因此 merged state 中的值就是當天零點到當前這一分鐘的累計值，我們的輸出結果就是從 merged state 得到的。
4.4.cumulate window 怎么解決 tumble window + early-fire 的問題
問題1：tumble window + early-fire 處理時間觸發的問題。
cumulate window 可以以事件時間推進進行觸發。
問題2：tumble window + early-fire retract 流問題。
cumulate window 是 append 流，自然沒有 retract 流的問題。
5.總結
本文主要介紹了 window tvf 實現的 cumulate window 聚合類指標的場景案例以及其運行原理：
介紹了周期內累計 PV，UV 是我們最常用的指標場景質疑。
在 tumble window + early-fire 或者 groupby + minibatch 計算周期內累計 PV，UV 存在各種問題是，誕生了 cumulate window 幫我們解決了這些問題，并以一個案例進行說明。
本文轉載自微信公眾號「大數據羊說」
