Apache Flink Time & Window 深度解析

作者：邱從賢

1、 Window & Time 介紹

Apache Flink（以下簡稱 Flink） 是一個天然支持無限流數(shù)據(jù)處理的分布式計算框架，在 Flink 中 Window 可以將無限流切分成有限流，是處理有限流的核心組件，現(xiàn)在 Flink 中 Window 可以是時間驅(qū)動的（Time Window），也可以是數(shù)據(jù)驅(qū)動的（Count Window）。

下面的代碼是在 Flink 中使用 Window 的兩個示例

cdn.xitu.io/2019/5/6/16a8ae89e3096f7e?w=724&h=209&f=jpeg&s=41903">

2、 Window API 使用

從第一部分我們已經(jīng)知道 Window 的一些基本概念，以及相關(guān) API，下面我們以一個實(shí)際例子來看看怎么使用 Window 相關(guān)的 API。

代碼來自 flink-examples

上面的例子中我們首先會對每條數(shù)據(jù)進(jìn)行時間抽取，然后進(jìn)行 keyby，接著依次調(diào)用 window()，evictor(), trigger() 以及 maxBy()。下面我們重點(diǎn)來看 window(), evictor() 和 trigger() 這幾個方法。

2.1 WindowAssigner, Evictor 以及 Trigger

window 方法接收的輸入是一個WindowAssigner， WindowAssigner 負(fù)責(zé)將每條輸入的數(shù)據(jù)分發(fā)到正確的 window 中（一條數(shù)據(jù)可能同時分發(fā)到多個 Window 中），F(xiàn)link 提供了幾種通用的 WindowAssigner：tumbling window(窗口間的元素?zé)o重復(fù)），sliding window（窗口間的元素可能重復(fù)），session window 以及 global window。如果需要自己定制數(shù)據(jù)分發(fā)策略，則可以實(shí)現(xiàn)一個 class，繼承自 WindowAssigner。

Tumbling Window

Sliding Window

Session Window

Global Window

evictor 主要用于做一些數(shù)據(jù)的自定義操作，可以在執(zhí)行用戶代碼之前，也可以在執(zhí)行用戶代碼之后，更詳細(xì)的描述可以參考 org.apache.flink.streaming.api.windowing.evictors.Evictor 的 evicBefore 和 evicAfter 兩個方法。Flink 提供了如下三種通用的 evictor：

• CountEvictor 保留指定數(shù)量的元素

• DeltaEvictor 通過執(zhí)行用戶給定的 DeltaFunction 以及預(yù)設(shè)的 threshold，判斷是否刪除一個元素。

• TimeEvictor設(shè)定一個閾值 interval，刪除所有不再 max_ts - interval 范圍內(nèi)的元素，其中 max_ts 是窗口內(nèi)時間戳的最大值。

evictor 是可選的方法，如果用戶不選擇，則默認(rèn)沒有。

trigger 用來判斷一個窗口是否需要被觸發(fā)，每個 WindowAssigner 都自帶一個默認(rèn)的 trigger，如果默認(rèn)的 trigger 不能滿足你的需求，則可以自定義一個類，繼承自 Trigger 即可，我們詳細(xì)描述下 Trigger 的接口以及含義：

• onElement() 每次往 window 增加一個元素的時候都會觸發(fā)

• onEventTime() 當(dāng) event-time timer 被觸發(fā)的時候會調(diào)用

• onProcessingTime() 當(dāng) processing-time timer 被觸發(fā)的時候會調(diào)用

• onMerge() 對兩個 trigger 的 state 進(jìn)行 merge 操作

• clear() window 銷毀的時候被調(diào)用

上面的接口中前三個會返回一個 TriggerResult，TriggerResult 有如下幾種可能的選擇：

• CONTINUE 不做任何事情

• FIRE 觸發(fā) window

• PURGE 清空整個 window 的元素并銷毀窗口

• FIRE_AND_PURGE 觸發(fā)窗口，然后銷毀窗口

2.2 Time & Watermark

了解完上面的內(nèi)容后，對于時間驅(qū)動的窗口，我們還有兩個概念需要澄清：Time 和 Watermark。

我們知道在分布式環(huán)境中 Time 是一個很重要的概念，在 Flink 中 Time 可以分為三種Event-Time，Processing-Time 以及 Ingestion-Time，三者的關(guān)系我們可以從下圖中得知：

Event Time、Ingestion Time、Processing Time

Event-Time 表示事件發(fā)生的時間，Processing-Time 則表示處理消息的時間（墻上時間），Ingestion-Time 表示進(jìn)入到系統(tǒng)的時間。

在 Flink 中我們可以通過下面的方式進(jìn)行 Time 類型的設(shè)置

env.setStreamTimeCharacteristic(TimeCharacteristic.ProcessingTime); // 設(shè)置使用 ProcessingTime

了解了 Time 之后，我們還需要知道 Watermark 相關(guān)的概念。

我們可以考慮一個這樣的例子：某 App 會記錄用戶的所有點(diǎn)擊行為，并回傳日志（在網(wǎng)絡(luò)不好的情況下，先保存在本地，延后回傳）。A 用戶在 11:02 對 App 進(jìn)行操作，B 用戶在 11:03 操作了 App，但是 A 用戶的網(wǎng)絡(luò)不太穩(wěn)定，回傳日志延遲了，導(dǎo)致我們在服務(wù)端先接受到 B 用戶 11:03 的消息，然后再接受到 A 用戶 11:02 的消息，消息亂序了。

那我們怎么保證基于 event-time 的窗口在銷毀的時候，已經(jīng)處理完了所有的數(shù)據(jù)呢？這就是 watermark 的功能所在。watermark 會攜帶一個單調(diào)遞增的時間戳 t，watermark(t) 表示所有時間戳不大于 t 的數(shù)據(jù)都已經(jīng)到來了，未來小于等于t的數(shù)據(jù)不會再來，因此可以放心地觸發(fā)和銷毀窗口了。下圖中給了一個亂序數(shù)據(jù)流中的 watermark 例子

2.3 遲到的數(shù)據(jù)

上面的 watermark 讓我們能夠應(yīng)對亂序的數(shù)據(jù)，但是真實(shí)世界中我們沒法得到一個完美的 watermark 數(shù)值 — 要么沒法獲取到，要么耗費(fèi)太大，因此實(shí)際工作中我們會使用近似 watermark — 生成 watermark(t) 之后，還有較小的概率接受到時間戳 t 之前的數(shù)據(jù)，在 Flink 中將這些數(shù)據(jù)定義為 “l(fā)ate elements”, 同樣我們可以在 window 中指定是允許延遲的最大時間（默認(rèn)為 0），可以使用下面的代碼進(jìn)行設(shè)置

設(shè)置allowedLateness 之后，遲來的數(shù)據(jù)同樣可以觸發(fā)窗口，進(jìn)行輸出，利用 Flink 的 side output 機(jī)制，我們可以獲取到這些遲到的數(shù)據(jù)，使用方式如下：

需要注意的是，設(shè)置了 allowedLateness 之后，遲到的數(shù)據(jù)也可能觸發(fā)窗口，對于 Session window 來說，可能會對窗口進(jìn)行合并，產(chǎn)生預(yù)期外的行為。

3 Window 內(nèi)部實(shí)現(xiàn)

在討論 Window 內(nèi)部實(shí)現(xiàn)的時候，我們再通過下圖回顧一下 Window 的生命周期

每條數(shù)據(jù)過來之后，會由 WindowAssigner 分配到對應(yīng)的 Window，當(dāng) Window 被觸發(fā)之后，會交給 Evictor（如果沒有設(shè)置 Evictor 則跳過），然后處理 UserFunction。其中 WindowAssigner，Trigger，Evictor 我們都在上面討論過，而 UserFunction 則是用戶編寫的代碼。

整個流程還有一個問題需要討論：Window 中的狀態(tài)存儲。我們知道 Flink 是支持 Exactly Once 處理語義的，那么 Window 中的狀態(tài)存儲和普通的狀態(tài)存儲又有什么不一樣的地方呢？

首先給出具體的答案：從接口上可以認(rèn)為沒有區(qū)別，但是每個 Window 會屬于不同的 namespace，而非 Window 場景下，則都屬于 VoidNamespace ，最終由 State/Checkpoint 來保證數(shù)據(jù)的 Exactly Once 語義，下面我們從 org.apache.flink.streaming.runtime.operators.windowing.WindowOperator 摘取一段代碼進(jìn)行闡述

從上面我們可以知道，Window 中的的元素同樣是通過 state 進(jìn)行維護(hù)，然后由 Checkpoint 機(jī)制保證 Exactly Once 語義。

至此，Time、Window 相關(guān)的所有內(nèi)容都已經(jīng)講解完畢，主要包括為什么要有 Window； Window 中的三個核心組件：WindowAssigner、Trigger 和 Evictor；Window 中怎么處理亂序數(shù)據(jù)，亂序數(shù)據(jù)是否允許延遲，以及怎么處理遲到的數(shù)據(jù)；最后我們梳理了整個 Window 的數(shù)據(jù)流程，以及 Window 中怎么保證 Exactly Once 語義。

更多資訊請?jiān)L問 Apache Flink 中文社區(qū)網(wǎng)站