Flink 水位線簡介

什么是水位線

在事件時間語義下，我們不依賴系統時間，而是基於數據自帶的時間戳去定義了一個時鍾，用來表示當前時間的進展。於是每個並行子任務都會有一個自己的邏輯時鍾，它的前進是靠數據的時間戳來驅動的。但在分布式系統中，這種驅動方式又會有一些問題。因為數據本身在處理轉換的過程中會變化，如果遇到窗口聚合這樣的操作，其實是要攢一批數據才會輸出一個結果，那么下游的數據就會變少，時間進度的控制就不夠精細了。另外，數據向下游任務傳遞時，一般只能傳輸給一個子任務（除廣播外），這樣其他的並行子任務的時鍾就無法推進了。例如一個時間戳為9整的數據到來，當前任務的時鍾就已經是9點了；處理完當前數據要發送到下游，如果下游任務是一個窗口計算，並行度為3，那么接收到這個數據的子任務，時鍾也會進展到9點，9點結束的窗口就可以關閉進行計算了；而另外兩個並行子任務則時間沒有變化，不能進行窗口計算。所以我們應該把時鍾也以數據的形式傳遞出去，告訴下游任務當前時間的進展；而且這個時鍾的傳遞不會因為窗口聚合之類的運算而停滯。一種簡單的想法是，在數據流中加入一個時鍾標記，記錄當前的事件時間；這個標記可以直接廣播到下游，當下游任務收到這個標記，就可以更新自己的時鍾了。由於類似於水流中用來做標志的記號，在Flink中，這種用來衡量事件時間（Event Time）進展的標記，就被稱作“水位線”（Watermark）。具體實現上，水位線可以看作一條特殊的數據記錄，它是插入到數據流中的一個標記點，主要內容就是一個時間戳，用來指示當前的事件時間。而它插入流中的位置，就應該是在某個數據到來之后；這樣就可以從這個數據中提取時間戳，作為當前水位線的時間戳了。

如上圖所示；每個事件產生的數據，都包含了一個時間戳，我們直接用一個整數表示。這里沒有指定單位，可以理解為秒或者毫秒（方便起見，下面講述統一認為是秒）。當產生於2秒的數據到來之后，當前的事件時間就是2秒；在后面插入一個時間戳也為2秒的水位線，隨着數據一起向下游流動。而當5秒產生的數據到來之后，同樣在后面插入一個水位線，時間戳也為5，當前的時鍾就推進到了5秒。這樣，如果出現下游有多個並行子任務的情形，我們只要將水位線廣播出去，就可以通知到所有下游任務當前的時間進度了。水位線就像它的名字所表達的，是數據流中的一部分，隨着數據一起流動，在不同任務之間傳輸。這看起來非常簡單；接下來我們就進一步探討一些復雜的狀況

1. 有序流中的水位線

在理想狀態下，數據應該按照它們生成的先后順序、排好隊進入流中；也就是說，它們處理的過程會保持原先的順序不變，遵守先來后到的原則。這樣的話我們從每個數據中提取時間戳，就可以保證總是從小到大增長的，從而插入的水位線也會不斷增長、事件時鍾不斷向前推進。實際應用中，如果當前數據量非常大，可能會有很多數據的時間戳是相同的，這時每來一條數據就提取時間戳、插入水位線就做了大量的無用功。而且即使時間戳不同，同時涌來的數據時間差會非常小（比如幾毫秒），往往對處理計算也沒什么影響。所以為了提高效率，一般會每隔一段時間生成一個水位線，這個水位線的時間戳，就是當前最新數據的時間戳。所以這時的水位線，其實就是有序流中的一個周期性出現的時間標記。

  

這里需要注意的是，水位線插入的“周期”，本身也是一個時間概念。在當前事件時間語義下，假如我們設定了每隔100ms生成一次水位線，那就是要等事件時鍾推進100ms才能插入；但是事件時鍾本身的進展，本身就是靠水位線來表示的——現在要插入一個水位線，可前提又是水位線要向前推進100ms，這就陷入了死循環。所以對於水位線的周期性生成，周期時間是指處理時間（系統時間），而不是事件時間。

2. 亂序流中的水位線

有序流的處理非常簡單，看起來水位線也並沒有起到太大的作用。但這種情況只存在於理想狀態下。我們知道在分布式系統中，數據在節點間傳輸，會因為網絡傳輸延遲的不確定性，導致順序發生改變，這就是所謂的“亂序數據”。這里所說的“亂序”（out-of-order），是指數據的先后順序不一致，主要就是基於數據的產生時間而言的。如下圖所示，一個7秒時產生的數據，生成時間自然要比9秒的數據早；但是經過數據緩存和傳輸之后，處理任務可能先收到了9秒的數據，之后7秒的數據才姍姍來遲。這時如果我們希望插入水位線，來指示當前的事件時間進展，又該怎么做呢？

最直觀的想法自然是跟之前一樣，我們還是靠數據來驅動，每來一個數據就提取它的時間戳、插入一個水位線。不過現在的情況是數據亂序，所以有可能新的時間戳比之前的還小，如果直接將這個時間的水位線再插入，我們的“時鍾”就回退了——水位線就代表了時鍾，時光不能倒流，所以水位線的時間戳也不能減小。解決思路也很簡單：我們插入新的水位線時，要先判斷一下時間戳是否比之前的大，否則就不再生成新的水位線，也就是說，只有數據的時間戳比當前時鍾大，才能推動時鍾前進，這時才插入水位線。

 

 如果考慮到大量數據同時到來的處理效率，我們同樣可以周期性地生成水位線。這時只需要保存一下之前所有數據中的最大時間戳，需要插入水位線時，就直接以它作為時間戳生成新的水位線

 

這樣做盡管可以定義出一個事件時鍾，卻也會帶來一個非常大的問題：我們無法正確處理“遲到”的數據。在上面的例子中，當9秒產生的數據到來之后，我們就直接將時鍾推進到了9秒；如果有一個窗口結束時間就是9秒（比如，要統計0~9秒的所有數據），那么這時窗口就應該關閉、將收集到的所有數據計算輸出結果了。但事實上，由於數據是亂序的，還可能有時間戳為7秒、8秒的數據在9秒的數據之后才到來，這就是“遲到數據”（late data）。它們本來也應該屬於0~9秒這個窗口，但此時窗口已經關閉，於是這些數據就被遺漏了，這會導致統計結果不正確。如果用之前我們類比班車的例子，現在的狀況就是商品不是按照生產時間順序到來的，所以有可能出現這種情況：9點生產的商品已經到了，我們認為已經到了9點，所以直接發車；但是可能還會有8點59分59秒生產的商品遲到了，沒有趕上這班車。那怎么解決這個問題呢？其實我們有很多生活中的經驗。假如是一個團隊出去團建，那肯定希望每個人都不能落下；如果有人因為堵車沒能准時到車上，我們可以稍微等一會兒。9點發車，我們可以等到9點10分，等人都到齊了再出發。當然，實際應用的網絡環境不可能跟北京的交通一樣堵，所以不需要等那么久，或許只要等一兩秒鍾就可以了。具體在商品班車的例子里，我們可以多等2秒鍾，也就是當生產時間為9點零2秒的商品到達，時鍾推進到9點零2秒，這時就認為所有8點到9點生產的商品都到齊了，可以正式發車。不過這樣相當於更改了發車時間，屬於“違規操作”。為了做到形式上仍然是9點發車，我們可以更改一下時鍾推進的邏輯：當一個商品到達時，不要直接用它的生產時間作為當前時間，而是減上兩秒，這就相當於把車上的邏輯時鍾調慢了。這樣一來，當9點生產的商品到達時，我們當前車上的時間是8點59分58秒，還沒到發車時間；當9點零2秒生產的商品到達時，車上時間剛好是9點，這時該到的商品都到齊了，准時發車就沒問題了。回到上面的例子，為了讓窗口能夠正確收集到遲到的數據，我們也可以等上2秒；也就是用當前已有數據的最大時間戳減去2秒，就是要插入的水位線的時間戳，如下圖所示。這樣的話，9秒的數據到來之后，事件時鍾不會直接推進到9秒，而是進展到了7秒；必須等到11秒的數據到來之后，事件時鍾才會進展到9秒，這時遲到數據也都已收集齊，0~9秒的窗口就可以正確計算結果了。

如果仔細觀察就會看到，這種“等2秒”的策略其實並不能處理所有的亂序數據。比如22秒的數據到來之后，插入的水位線時間戳為20，也就是當前時鍾已經推進到了20秒；對於10~20秒的窗口，這時就該關閉了。但是之后又會有17秒的遲到數據到來，它本來應該屬於10~20秒窗口，現在卻被遺漏丟棄了。那又該怎么辦呢？既然現在等2秒還是等不到17秒產生的遲到數據，那自然我們可以試着多等幾秒，也就是把時鍾調得更慢一些。最終的目的，就是要讓窗口能夠把所有遲到數據都收進來，得到正確的計算結果。對應到水位線上，其實就是要保證，當前時間已經進展到了這個時間戳，在這之后不可能再有遲到數據來了。下面是一個示例，我們可以使用周期性的方式生成正確的水位線。

第一個水位線時間戳為7，它表示當前事件時間是7秒，7秒之前的數據都已經到齊，之后再也不會有了；同樣，第二個、第三個水位線時間戳分別為12和20，表示11秒、20秒之前的數據都已經到齊，如果有對應的窗口就可以直接關閉了，統計的結果一定是正確的。這里由於水位線是周期性生成的，所以插入的位置不一定是在時間戳最大的數據后面。另外需要注意的是，這里一個窗口所收集的數據，並不是之前所有已經到達的數據。因為數據屬於哪個窗口，是由數據本身的時間戳決定的，一個窗口只會收集真正屬於它的那些數據。也就是說，上圖中盡管水位線W(20)之前有時間戳為22的數據到來，10~20秒的窗口中也不會收集這個數據，進行計算依然可以得到正確的結果。關於窗口的原理，我們會在后面繼續展開講解。

水位線特點

• 水位線是插入到數據流中的一個標記，可以認為是一個特殊的數據
• 水位線主要的內容是一個時間戳，用來表示當前事件時間的進展
• 水位線是基於數據的時間戳生成的
• 水位線的時間戳必須單調遞增，以確保任務的事件時間時鍾一直向前推進
• 水位線可以通過設置延遲，來保證正確處理亂序數據
• 一個水位線Watermark(t)，表示在當前流中事件時間已經達到了時間戳t, 這代表t之前的所有數據都到齊了，之后流中不會出現時間戳t’≤t的數據