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

一、Flink的三種時間語義

圖片

1.1 Event Time

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

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

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

使用Event Time的優(yōu)勢是結(jié)果的可預(yù)測性，缺點是緩存較大，增加了延遲，且調(diào)試和定位問題更復(fù)雜。

1.2 Processing Time

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

Processing Time只依賴當(dāng)前執(zhí)行機(jī)器的系統(tǒng)時鐘，不需要依賴Watermark，無需緩存。Processing Time是實現(xiàn)起來非常簡單也是延遲最小的一種時間語義。

1.3 Ingestion Time

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

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

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

1.4 Flink如何設(shè)置時間域

調(diào)用 setStreamTimeCharacteristic 設(shè)置時間域，枚舉類 TimeCharacteristic 預(yù)設(shè)了三種時間域，不顯式設(shè)置的情況下，默認(rèn)使用 TimeCharacteristic.EventTime（1.12 版本以前默認(rèn)是 TimeCharacteristic.ProcessingTime）。

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

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

env.getConfig().setAutoWatermarkInterval(0);

二、Flink 水印機(jī)制

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

稍稍總結(jié)一下水位線的引入原因：

• 分布式系統(tǒng)的網(wǎng)絡(luò)傳輸?shù)牟淮_定性；
• 數(shù)據(jù)是亂序的；
• 支持事件時間的流處理器需要一種測量事件時間進(jìn)度的方法，用以正確的處理窗口等操作；

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

• 水位線本質(zhì)是一個基于數(shù)據(jù)生成的、單調(diào)遞增的時間戳；
• 水位線 W(t)表示當(dāng)前數(shù)據(jù)流中的所有 t 時刻前的數(shù)據(jù)都已經(jīng)到了。

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

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

圖片

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

圖片

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

圖片

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

圖片

三、分布式環(huán)境下Watermark的傳播

在實際計算過程中，F(xiàn)link的算子一般分布在多個并行的分區(qū)（或者稱為實例）上，F(xiàn)link需要將Watermark在并行環(huán)境下向前傳播。如下圖所示，F(xiàn)link的每個并行算子子任務(wù)會維護(hù)針對該子任務(wù)的Event Time時鐘，這個時鐘記錄了這個算子子任務(wù)Watermark處理進(jìn)度，隨著上游Watermark數(shù)據(jù)不斷向下發(fā)送，算子子任務(wù)的Event Time時鐘也要不斷向前更新。由于上游各分區(qū)的處理速度不同，到達(dá)當(dāng)前算子的Watermark也會有先后快慢之分，每個算子子任務(wù)會維護(hù)來自上游不同分區(qū)的Watermark信息，這是一個列表，列表內(nèi)對應(yīng)上游算子各分區(qū)的Watermark時間戳等信息。

圖片

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

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

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

此外，在union等多數(shù)據(jù)流處理時，F(xiàn)link也使用上述Watermark更新機(jī)制，那就意味著，多個數(shù)據(jù)流的時間必須對齊，如果一方的Watermark時間較老，那整個應(yīng)用的Event Time時鐘也會使用這個較老的時間，其他數(shù)據(jù)流的數(shù)據(jù)會被積壓。一旦發(fā)現(xiàn)某個數(shù)據(jù)流不再生成新的Watermark，我們要在SourceFunction中的SourceContext里調(diào)用markAsTemporarilyIdle設(shè)置該數(shù)據(jù)流為空閑狀態(tài)。

四、Flink對亂序數(shù)據(jù)的三重保障

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

常用的解決辦法是：當(dāng)最大的事件時間maxEventTime達(dá)到了窗口關(guān)閉時間，不應(yīng)該立刻觸發(fā)窗口計算，而是等待一段時間，等遲到的數(shù)據(jù)來了再關(guān)閉窗口。

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

處理亂序數(shù)據(jù)，三重保證機(jī)制：

3.1 Watermark

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

3.2 allowedLateness

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

3.3 sideOutputLateData

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

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

注意：

后期觸發(fā)的元素應(yīng)更新先前計算的結(jié)果，即數(shù)據(jù)流將包含同一計算的多個結(jié)果。根據(jù)你的應(yīng)用程序，需要考慮這些重復(fù)的結(jié)果或?qū)λ鼈冞M(jìn)行重復(fù)數(shù)據(jù)刪除。

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

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

圖片

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

圖片

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

圖片

當(dāng)數(shù)據(jù)A到達(dá)時，由于窗口開始時間<=數(shù)據(jù)A的事件時間<窗口結(jié)束時間，因此數(shù)據(jù)A落入窗口內(nèi)。

當(dāng)數(shù)據(jù)B到達(dá)時，由于其事件時間>=窗口結(jié)束時間，因此數(shù)據(jù)B不屬于該窗口。此時Watermark=進(jìn)入Flink的當(dāng)前最大事件時間?允許的最大延遲時間=9:11?3分鐘=9:08。水印在窗口內(nèi)，不會觸發(fā)窗口計算。

當(dāng)數(shù)據(jù)C到達(dá)時，由于窗口開始時間<=數(shù)據(jù)C的事件時間<窗口結(jié)束時間，因此數(shù)據(jù)C落入窗口內(nèi)。

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

當(dāng)數(shù)據(jù)E到達(dá)時，由于進(jìn)入Flink的當(dāng)前最大事件時間沒有改變，窗口不會關(guān)閉，而是繼續(xù)等待。窗口開始時間<=數(shù)據(jù)E的事件時間<窗口結(jié)束時間，因此數(shù)據(jù)E落入窗口內(nèi)，并觸發(fā)窗口計算，與上次計算的結(jié)果進(jìn)行合并，發(fā)射出新的計算結(jié)果，如下圖所示。

圖片

當(dāng)數(shù)據(jù)F到達(dá)時，此時的窗口結(jié)束時間+允許的最大延遲時間（水印指定的）+允許延遲機(jī)制指定的延遲時間=9:10+3（分鐘）+3（分鐘）=9:16<=9:16（進(jìn)入Flink的當(dāng)前最大事件時間），滿足窗口關(guān)閉的條件，因此窗口會關(guān)閉并銷毀。

當(dāng)數(shù)據(jù)G到達(dá)時，窗口開始時間<=數(shù)據(jù)G的事件時間<窗口結(jié)束時間，但是窗口已經(jīng)關(guān)閉了，因此數(shù)據(jù)G將進(jìn)入側(cè)道輸出流進(jìn)行單獨存放。通過側(cè)道輸出API可從側(cè)道輸出流中取出延遲嚴(yán)重的數(shù)據(jù)進(jìn)行相應(yīng)的業(yè)務(wù)處理。