Apache Flink Time & Window 深度解析

作者：邱從賢

1、 Window & Time 介紹

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

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

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

2、 Window API 使用

從第一部分我們已經知道 Window 的一些基本概念，以及相關 API，下面我們以一個實際例子來看看怎么使用 Window 相關的 API。

代碼來自 flink-examples

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

2.1 WindowAssigner, Evictor 以及 Trigger

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

Tumbling Window

Sliding Window

Session Window

Global Window

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

• CountEvictor 保留指定數量的元素

• DeltaEvictor 通過執行用戶給定的 DeltaFunction 以及預設的 threshold，判斷是否刪除一個元素。

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

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

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

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

• onEventTime() 當 event-time timer 被觸發的時候會調用

• onProcessingTime() 當 processing-time timer 被觸發的時候會調用

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

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

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

• CONTINUE 不做任何事情

• FIRE 觸發 window

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

• FIRE_AND_PURGE 觸發窗口，然后銷毀窗口

2.2 Time & Watermark

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

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

Event Time、Ingestion Time、Processing Time

Event-Time 表示事件發生的時間，Processing-Time 則表示處理消息的時間（墻上時間），Ingestion-Time 表示進入到系統的時間。

在 Flink 中我們可以通過下面的方式進行 Time 類型的設置

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

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

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

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

2.3 遲到的數據

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

設置allowedLateness 之后，遲來的數據同樣可以觸發窗口，進行輸出，利用 Flink 的 side output 機制，我們可以獲取到這些遲到的數據，使用方式如下：

需要注意的是，設置了 allowedLateness 之后，遲到的數據也可能觸發窗口，對于 Session window 來說，可能會對窗口進行合并，產生預期外的行為。

3 Window 內部實現

在討論 Window 內部實現的時候，我們再通過下圖回顧一下 Window 的生命周期

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

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

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

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

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

更多資訊請訪問 Apache Flink 中文社區網站