Flink SQL知其所以然：大家都用 Cumulate Window 計(jì)算累計(jì)指標(biāo)啦

作者：antigeneral了呀 2021-11-24 08:17:21

本文主要介紹了 window tvf 實(shí)現(xiàn)的 cumulate window 聚合類指標(biāo)的場景案例以及其運(yùn)行原理,介紹了周期內(nèi)累計(jì) PV，UV 是我們最常用的指標(biāo)場景質(zhì)疑。

[[436466]]

1.序篇

此節(jié)就是窗口聚合章節(jié)的第三篇，上節(jié)介紹了 1.13 window tvf tumble window 實(shí)現(xiàn)，本節(jié)主要介紹 1.13. window tvf 的一個(gè)重磅更新，即 cumulate window。

本節(jié)從以下幾個(gè)章節(jié)給大家詳細(xì)介紹 cumulate window 的能力。

應(yīng)用場景介紹
預(yù)期的效果
解決方案介紹
總結(jié)及展望篇

2.應(yīng)用場景介紹

先來一個(gè)簡單的小調(diào)查：在實(shí)時(shí)場景中，你見到過最多的指標(biāo)需求場景是哪一種?

答案：博主相信，占比比較多的不是 PCU(即同時(shí)在線 PV，UV)，而是周期內(nèi)累計(jì) PV，UV 指標(biāo)(如每天累計(jì)到當(dāng)前這一分鐘的 PV，UV)。因?yàn)檫@類指標(biāo)是一段周期內(nèi)的累計(jì)狀態(tài)，對(duì)分析師來說更具統(tǒng)計(jì)分析價(jià)值，而且?guī)缀跛械膹?fù)合指標(biāo)都是基于此類指標(biāo)的統(tǒng)計(jì)(不然離線為啥都要一天的數(shù)據(jù)，而不要一分鐘的數(shù)據(jù)呢)。

本文要介紹的就是周期內(nèi)累計(jì) PV，UV 指標(biāo)在 flink 1.13 版本的最優(yōu)解決方案。

3.預(yù)期的效果

先來一個(gè)實(shí)際案例來看看在具體輸入值的場景下，輸出值應(yīng)該長啥樣。

指標(biāo)：每天的截止當(dāng)前分鐘的累計(jì) money(sum(money))，去重 id 數(shù)(count(distinct id))。每天代表窗口大小為 1 天，分鐘代表移動(dòng)步長為分鐘級(jí)別。

來一波輸入數(shù)據(jù)：

預(yù)期輸出數(shù)據(jù)：

轉(zhuǎn)化為折線圖長這樣：

當(dāng)日累計(jì)

可以看到，其特點(diǎn)就在于，每一分鐘的輸出結(jié)果都是當(dāng)天零點(diǎn)累計(jì)到當(dāng)前的結(jié)果。

4.解決方案介紹

4.1.flink 1.13 之前

可選的解決方案有兩種

tumble window(1天窗口) + early-fire(1分鐘)
group by(1天) + minibatch(1分鐘)

但是上述兩種解決方案產(chǎn)出的都是 retract 流，關(guān)于 retract 流存在的缺點(diǎn)見如下文章：

踩坑記 | flink sql count 還有這種坑!

并且 tumble window + early-fire 的觸發(fā)機(jī)制是基于處理時(shí)間而非事件時(shí)間，具體缺點(diǎn)見如下文章：

https://mp.weixin.qq.com/s/L8-RSS6v3Ppts60CWngiOA

4.2.flink 1.13 及之后

誕生了 cumulate window 解法，具體見官網(wǎng)鏈接：

https://nightlies.apache.org/flink/flink-docs-release-1.13/docs/dev/table/sql/queries/window-tvf/#cumulate

如下官網(wǎng)文檔所示，介紹 cumulate window 的第一句話就是 cumulate window 非常適合于之前使用 tumble window + early-fire 的場景?？梢哉f cumulate window 就是在用戶計(jì)算周期內(nèi)累計(jì) PV，UV 指標(biāo)時(shí)，使用了 tumble window + early-fire 后發(fā)現(xiàn)這種方案存在了很多坑的情況下，而誕生的!

cumulate window

其計(jì)算機(jī)制如下圖所示：

cumulate window

還是以剛剛的案例說明，以天為窗口，每分鐘輸出一次當(dāng)天零點(diǎn)到當(dāng)前分鐘的累計(jì)值，在 cumulate window 中，其窗口劃分規(guī)則如下：

[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]

第一個(gè) window 統(tǒng)計(jì)的是一個(gè)區(qū)間的數(shù)據(jù);第二個(gè) window 統(tǒng)計(jì)的是第一區(qū)間和第二個(gè)區(qū)間的數(shù)據(jù);第三個(gè) window 統(tǒng)計(jì)的是第一區(qū)間，第二個(gè)區(qū)間和第三個(gè)區(qū)間的數(shù)據(jù)。

那么以 cumulate window 實(shí)現(xiàn)上述的需求，具體的 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 自動(dòng)生成的類型是 timestamp(3)。

window_start 固定為窗口的開始時(shí)間。window_end 為一個(gè)子窗口的結(jié)束時(shí)間。

最終結(jié)果如下。

輸入數(shù)據(jù)：

輸出數(shù)據(jù)：

Notes：天級(jí)別窗口劃分的時(shí)候一定要注意時(shí)區(qū)問題喔!https://nightlies.apache.org/flink/flink-docs-master/zh/docs/dev/table/timezone/

4.3.cumulate window 原理解析

首先 cumulate window 是一個(gè)窗口，其窗口計(jì)算的觸發(fā)也是完全由 watermark 推動(dòng)的。與 tumble window 一樣。

以上述天窗口分鐘累計(jì)案例舉例：cumulate window 維護(hù)了一個(gè) slice state 和 merged state，slice state 就是每一分鐘內(nèi)窗口數(shù)據(jù)(叫做切片)，merged state 的作用是當(dāng) watermark 推動(dòng)到下一分鐘時(shí)，這一分鐘的 slice state 就會(huì)被 merge 到 merged stated 中，因此 merged state 中的值就是當(dāng)天零點(diǎn)到當(dāng)前這一分鐘的累計(jì)值，我們的輸出結(jié)果就是從 merged state 得到的。