[[436466]]

1.序篇

此節就是窗口聚合章節的第三篇，上節介紹了 1.13 window tvf tumble window 實現，本節主要介紹 1.13. window tvf 的一個重磅更新，即 cumulate window。

本節從以下幾個章節給大家詳細介紹 cumulate window 的能力。

1. 應用場景介紹
2. 預期的效果
3. 解決方案介紹
4. 總結及展望篇

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 之前

可選的解決方案有兩種

1. tumble window(1天窗口) + early-fire(1分鐘)
2. 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 的場景。可以說 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 幫我們解決了這些問題，并以一個案例進行說明。

本文轉載自微信公眾號「大數據羊說」