該篇主要講Flink中的時間語義、Flink 水印機制以及Flink對亂序數據的三重保障。

一、Flink的三種時間語義

圖片

1.1 Event Time

Event Time指的是數據流中每個元素或者每個事件自帶的時間屬性，一般是事件發生的時間。由于事件從發生到進入Flink時間算子之間有很多環節，一個較早發生的事件因為延遲可能較晚到達，因此使用Event Time意味著事件到達有可能是亂序的。

使用Event Time時，最理想的情況下，我們可以一直等待所有的事件到達后再進行時間窗口的處理。假設一個時間窗口內的所有數據都已經到達，基于Event Time的流處理會得到正確且一致的結果：無論我們是將同一個程序部署在不同的計算環境還是在相同的環境下多次計算同一份數據，都能夠得到同樣的計算結果。我們根本不同擔心亂序到達的問題。但這只是理想情況，現實中無法實現，因為我們既不知道究竟要等多長時間才能確認所有事件都已經到達，更不可能無限地一直等待下去。在實際應用中，當涉及到對事件按照時間窗口進行統計時，Flink會將窗口內的事件緩存下來，直到接收到一個Watermark，以確認不會有更晚數據的到達。Watermark意味著在一個時間窗口下，Flink會等待一個有限的時間，這在一定程度上降低了計算結果的絕對準確性，而且增加了系統的延遲。在流處理領域，比起其他幾種時間語義，使用Event Time的好處是某個事件的時間是確定的，這樣能夠保證計算結果在一定程度上的可預測性。

一個基于Event Time的Flink程序中必須定義Event Time，以及如何生成Watermark。我們可以使用元素中自帶的時間，也可以在元素到達Flink后人為給Event Time賦值。

使用Event Time的優勢是結果的可預測性，缺點是緩存較大，增加了延遲，且調試和定位問題更復雜。

1.2 Processing Time

對于某個算子來說，Processing Time指算子使用當前機器的系統時鐘來定義時間。在Processing Time的時間窗口場景下，無論事件什么時候發生，只要該事件在某個時間段達到了某個算子，就會被歸結到該窗口下，不需要Watermark機制。對于一個程序在同一個計算環境來說，每個算子都有一定的耗時，同一個事件的Processing Time，第n個算子和第n+1個算子不同。如果一個程序在不同的集群和環境下執行時，限于軟硬件因素，不同環境下前序算子處理速度不同，對于下游算子來說，事件的Processing Time也會不同，不同環境下時間窗口的計算結果會發生變化。因此，Processing Time在時間窗口下的計算會有不確定性。

Processing Time只依賴當前執行機器的系統時鐘，不需要依賴Watermark，無需緩存。Processing Time是實現起來非常簡單也是延遲最小的一種時間語義。

1.3 Ingestion Time

Ingestion Time是事件到達Flink Souce的時間。從Source到下游各個算子中間可能有很多計算環節，任何一個算子的處理速度快慢可能影響到下游算子的Processing Time。而Ingestion Time定義的是數據流最早進入Flink的時間，因此不會被算子處理速度影響。

Ingestion Time通常是Event Time和Processing Time之間的一個折中方案。比起Event Time，Ingestion Time可以不需要設置復雜的Watermark，因此也不需要太多緩存，延遲較低。比起Processing Time，Ingestion Time的時間是Souce賦值的，一個事件在整個處理過程從頭至尾都使用這個時間，而且后續算子不受前序算子處理速度的影響，計算結果相對準確一些，但計算成本稍高。

注：Ingestion Time1.13 版本已經不再提了，這也是為啥官網的圖沒看到Ingestion Time的原因。目前推薦Event Time的時間語義。

1.4 Flink如何設置時間域

調用 setStreamTimeCharacteristic 設置時間域，枚舉類 TimeCharacteristic 預設了三種時間域，不顯式設置的情況下，默認使用 TimeCharacteristic.EventTime（1.12 版本以前默認是 TimeCharacteristic.ProcessingTime）。

env = StreamExecutionEnvironment.getExecutionEnvironment();
env.setStreamTimeCharacteristic(TimeCharacteristic.ProcessingTime); //過期方法

在 1.12 以后版本默認是使用 EventTime，如果要顯示使用 ProcessingTime，可以關閉 watermark（自動生成 watermark 的間隔設置為 0），設置

env.getConfig().setAutoWatermarkInterval(0);

二、Flink 水印機制

我們知道，流處理從事件產生，到流經 source，再到 operator，中間是有一個過程和時間的，雖然大部分情況下，流到 operator 的數據都是按照事件產生的時間順序來的，但是也不排除由于網絡、分布式等原因，導致亂序的產生，所謂亂序，就是指 Flink 接收到的事件的先后順序不是嚴格按照事件的 Event Time 順序排列的，為了保證計算結果的正確性，需要讓窗口等待延遲數據到達后再進行計算，但是不能無限期地等待下去，必須有一種機制來確定何時觸發窗口計算，這種機制就是水印(Watermark)。

稍稍總結一下水位線的引入原因：

• 分布式系統的網絡傳輸的不確定性；
• 數據是亂序的；
• 支持事件時間的流處理器需要一種測量事件時間進度的方法，用以正確的處理窗口等操作；

水位線的物理意義有兩點：

• 水位線本質是一個基于數據生成的、單調遞增的時間戳；
• 水位線 W(t)表示當前數據流中的所有 t 時刻前的數據都已經到了。

水印是一種用于衡量事件時間進度的機制，其表示某個時刻（事件時間）以前的數據將不再產生，因此水印指的是一個時間點。水印作為數據流的一部分流動，并帶有時間戳t。t表示該流中不應再有時間戳小于等于t的元素（即時間戳早于或等于水印的事件）。

下圖顯示了帶有時間戳和嵌入式水印的事件流，事件是按順序排列的（相對于其時間戳），這意味著水印只是流中的周期性標記。

圖片

水印對于亂序流至關重要，如下圖所示，其中事件不是按其時間戳排序的。通常，水印是數據流中一個點的聲明，表示水印之前的所有事件都應該到達。一旦水印到達算子，算子則認為某個時間周期，所有事件已經被收到，不會再有更多符合條件的事件。

圖片

水印是直接通過Source Function生成的或在后續的DataStream API中生成的。在實際的流計算中，一個作業往往會同時處理多個源的數據，多個源的數據按照key分組后進行Shuffle處理，數據會匯聚到同一個處理節點。而每個并行子任務通常獨立生成水印，這樣就容易導致匯聚到一起的水印不是單調遞增的。對于這種情況，Flink會選擇所有流入的水印中事件時間最小的一個發往下游，如下圖所示。

圖片

多個流的水印流入算子后，由于當前算子也有自己的水印，因此算子會綜合計算得出最終水印，計算規則為：取多個流中事件時間最小的水印與當前算子的水印進行對比，如果大于當前算子水印，則更新當前算子水印，并發往下游。例如抽象類AbstractStreamOperator中的源碼如下：

圖片

三、分布式環境下Watermark的傳播

在實際計算過程中，Flink的算子一般分布在多個并行的分區（或者稱為實例）上，Flink需要將Watermark在并行環境下向前傳播。如下圖所示，Flink的每個并行算子子任務會維護針對該子任務的Event Time時鐘，這個時鐘記錄了這個算子子任務Watermark處理進度，隨著上游Watermark數據不斷向下發送，算子子任務的Event Time時鐘也要不斷向前更新。由于上游各分區的處理速度不同，到達當前算子的Watermark也會有先后快慢之分，每個算子子任務會維護來自上游不同分區的Watermark信息，這是一個列表，列表內對應上游算子各分區的Watermark時間戳等信息。

圖片

當上游某分區有Watermark進入該算子子任務后，Flink先判斷新流入的Watermark時間戳是否大于Partition Watermark列表內記錄的該分區的歷史Watermark時間戳，如果新流入的更大，則更新該分區的Watermark。例如，某個分區新流入的Watermark時間戳為4，算子子任務維護的該分區Watermark為1，那么Flink會更新Partition Watermark列表為最新的時間戳4。接著，Flink會遍歷Partition Watermark列表中的所有時間戳，選擇最小的一個作為該算子子任務的Event Time。同時，Flink會將更新的Event Time作為Watermark發送給下游所有算子子任務。算子子任務Event Time的更新意味著該子任務將時間推進到了這個時間，該時間之前的事件已經被處理并發送到下游。例如，圖中第二步和第三步，Partition Watermark列表更新后，導致列表中最小時間戳發生了變化，算子子任務的Event Time時鐘也相應進行了更新。整個過程完成了數據流中的Watermark推動算子子任務Watermark的時鐘更新過程。Watermark像一個幕后推動者，不斷將流處理系統的Event Time向前推進。我們可以將這種機制總結為：

• Flink某算子子任務根據各上游流入的Watermark來更新Partition Watermark列表。
• 選取Partition Watermark列表中最小的時間作為該算子的Event Time，并將這個時間發送給下游算子。

這樣的設計機制滿足了并行環境下Watermark在各算子中的傳播問題，但是假如某個上游分區的Watermark一直不更新，Partition Watermark列表其他地方都在正常更新，唯獨個別分區的時間停留在很早的某個時間，這會導致算子的Event Time時鐘不更新，相應的時間窗口計算也不會被觸發，大量的數據積壓在算子內部得不到處理，整個流處理處于空轉狀態。這種問題可能出現在使用數據流自帶的Watermark，自帶的Watermark在某些分區下沒有及時更新。針對這種問題，一種解決辦法是根據機器當前的時鐘周期性地生成Watermark。

此外，在union等多數據流處理時，Flink也使用上述Watermark更新機制，那就意味著，多個數據流的時間必須對齊，如果一方的Watermark時間較老，那整個應用的Event Time時鐘也會使用這個較老的時間，其他數據流的數據會被積壓。一旦發現某個數據流不再生成新的Watermark，我們要在SourceFunction中的SourceContext里調用markAsTemporarilyIdle設置該數據流為空閑狀態。

四、Flink對亂序數據的三重保障

我們思考一個問題：怎樣避免亂序數據帶來計算不正確性？

常用的解決辦法是：當最大的事件時間maxEventTime達到了窗口關閉時間，不應該立刻觸發窗口計算，而是等待一段時間，等遲到的數據來了再關閉窗口。

但是，我們應該等待多久的時間呢？由于網絡、分布式等原因造成的延時，一般大多數遲到的數據都會在最近一段時間到來，這個最近一段時間一般是毫秒級的，Watermark就是做到了這樣的保障。還有很少的一部分數據會遲到很久，我們可以通過allowedLateness和sideOutputLateData來兜底。

處理亂序數據，三重保證機制：

3.1 Watermark

能夠保證遲到很短的時間的數據到來后（一般是遲到毫秒級別內的數據，最大不超過1s），觸發窗口關閉并輸出。（即能夠hold住短時間內遲到的數據）

3.2 allowedLateness

allowedLateness(lateness: Time)：設置允許的延遲時間，默認為0，該方法僅對事件時間窗口有效。在水印通過窗口結尾后（即水印>=窗口結束時間），該方法指定的允許延遲時間才開始生效。該延遲時間與水印指定的允許延遲時間不沖突，相當于在水印延遲時間的基礎上進行累加。落入該方法指定的允許延遲時間范圍內的元素可能會導致窗口再次觸發（例如EventTimeTrigger）。為了使這些元素正常被計算，Flink會保持窗口的狀態，直到允許的延遲過期為止。一旦延遲過期，Flink將刪除該窗口并刪除其狀態。

3.3 sideOutputLateData

sideOutputLateData(outputTag: OutputTag[T])：將延遲到達的數據保存到outputTag對象中，OutputTag是一種類型化的命名標簽，用于標記算子的側道輸出，單獨收集延遲數據。后面可通過DataStream的getSideOutput(outputTag)方法得到被丟棄數據組成的數據流。

當指定的允許延遲大于0時，在水印通過窗口結尾后，將保留窗口及其內容。在這種情況下，當一個遲到但未被丟棄的元素到達時，它可能會導致該窗口的另一次觸發。這次觸發稱為延遲觸發，因為是由延遲事件觸發的，與主觸發（即窗口的第一次觸發）相反。對于會話窗口，后期觸發會進一步導致窗口合并，因為可能縮小兩個預先存在的未合并窗口之間的間隙。當使用全局窗口時，沒有數據是延遲的，因為全局窗口的結束時間戳是Long.MAX_VALUE。

注意：

后期觸發的元素應更新先前計算的結果，即數據流將包含同一計算的多個結果。根據你的應用程序，需要考慮這些重復的結果或對它們進行重復數據刪除。

在水印的基礎上設置允許延遲機制后，數據可以延遲的時間范圍是多少？在只設置了水印的情況下，如果滿足當前進入Flink的最大事件時間>=窗口結束時間+允許的最大延遲時間，則觸發窗口計算，發射計算結果并銷毀窗口。在水印的基礎上設置了允許延遲機制后，如果滿足當前進入Flink的最大事件時間>=窗口結束時間+允許的最大延遲時間（水印指定的），則觸發窗口計算，發射計算結果，但不會銷毀窗口，窗口會保留計算狀態并繼續等待延遲數據；每條延遲數據到達后，如果落入窗口內，都會再次觸發窗口計算，更新計算狀態，發射出最新計算結果，直到滿足條件：當前進入Flink的最大事件時間>=窗口結束時間+允許的最大延遲時間（水印指定的）+允許延遲機制指定的延遲時間，則關閉并銷毀窗口。此后到達的延遲數據，由于窗口已經關閉，數據將進入側道輸出流進行單獨存放，后期根據業務單獨處理即可。

指定允許延遲時間可以使用如下代碼片段：

圖片

使用Flink的側道輸出機制可以獲得一個后來被丟棄的數據組成的數據流。使用時首先需要使用sideOutputLateData(OutputTag)方法指定要在窗口化流上獲取后期數據。然后可以使用getSideOutput(lateOutputTag)方法得到后期數據組成的數據流，代碼如下：

圖片

為了更好地理解允許延遲和側道輸出機制，假設有亂序數據按照ABCDEFG的順序依次到達Flink應用程序，并且設置了水印允許的最大延遲時間為3分鐘，在水印的基礎上又通過allowedLateness(Time.minutes(3))方法設置了允許的延遲時間為3分鐘，使用sideOutputLateData(lateOutputTag)方法設置側道輸出，如下圖所示。

圖片

當數據A到達時，由于窗口開始時間<=數據A的事件時間<窗口結束時間，因此數據A落入窗口內。

當數據B到達時，由于其事件時間>=窗口結束時間，因此數據B不屬于該窗口。此時Watermark=進入Flink的當前最大事件時間?允許的最大延遲時間=9:11?3分鐘=9:08。水印在窗口內，不會觸發窗口計算。

當數據C到達時，由于窗口開始時間<=數據C的事件時間<窗口結束時間，因此數據C落入窗口內。

當數據D到達時，由于其事件時間>=窗口結束時間，因此數據D不屬于該窗口。此時Watermark=進入Flink的當前最大事件時間?允許的最大延遲時間=9:15?3（分鐘）=9:12>=窗口結束時間。水印在窗口外，觸發窗口計算并發射計算結果。由于設置了允許延遲機制的延遲時間為3分鐘，此時的窗口結束時間+允許的最大延遲時間（水印指定的）+允許延遲機制指定的延遲時間=9:10+3（分鐘）+3（分鐘）=9:16>9:15（進入Flink的當前最大事件時間），不滿足窗口關閉的條件，因此窗口會繼續等待延遲數據，并保留計算狀態（此處的計算狀態指的就是計算結果，例如窗口內數據的聚合結果）。

當數據E到達時，由于進入Flink的當前最大事件時間沒有改變，窗口不會關閉，而是繼續等待。窗口開始時間<=數據E的事件時間<窗口結束時間，因此數據E落入窗口內，并觸發窗口計算，與上次計算的結果進行合并，發射出新的計算結果，如下圖所示。

圖片

當數據F到達時，此時的窗口結束時間+允許的最大延遲時間（水印指定的）+允許延遲機制指定的延遲時間=9:10+3（分鐘）+3（分鐘）=9:16<=9:16（進入Flink的當前最大事件時間），滿足窗口關閉的條件，因此窗口會關閉并銷毀。

當數據G到達時，窗口開始時間<=數據G的事件時間<窗口結束時間，但是窗口已經關閉了，因此數據G將進入側道輸出流進行單獨存放。通過側道輸出API可從側道輸出流中取出延遲嚴重的數據進行相應的業務處理。