Flink流處理的時間視窗

對於流處理系統來說，流入的訊息是無限的，所以對於聚合或是連線等操作，流處理系統需要對流入的訊息進行分段，然後基於每一段資料進行聚合或是連線等操作。

訊息的分段即稱為視窗，流處理系統支援的視窗有很多型別，最常見的就是時間視窗，基於時間間隔對訊息進行分段處理。本節主要介紹Flink流處理系統支援的各種時間視窗。

對於目前大部分流處理系統來說，時間視窗一般是根據Task所在節點的本地時鐘來進行切分，這種方式實現起來比較容易，不會阻塞訊息處理。但是可能無法滿足某些應用的要求，例如：

1. 訊息本身帶有時間戳，使用者希望按照訊息本身的時間特性進行分段處理。

2. 由於不同節點的時鐘可能不同，以及訊息在流經各個節點時延遲不同，在某個節點屬於同一個時間視窗處理的訊息，流到下一個節點時可能被切分到不同的時間視窗中，從而產生不符合預期的結果。

Flink支援三種類型的時間視窗，分別適用於使用者對於時間視窗不同型別的要求：

1. Operator Time。根據Task所在節點的本地時鐘來進行切分的時間視窗。

2. Event Time。訊息自帶時間戳，根據訊息的時間戳進行處理，確保時間戳在同一個時間視窗的所有訊息一定會被正確處理。由於訊息可能是亂序流入Task的，

所以Task需要快取當前時間視窗訊息處理的狀態，直到確認屬於該時間視窗的所有訊息都被處理後，才可以釋放其狀態。如果亂序的訊息延遲很高的話，會影響分散式系統的吞吐量和延遲。

3. Ingress Time。有時訊息本身並不帶有時間戳資訊，但使用者依然希望按照訊息而不是節點時鐘劃分時間視窗(例如，避免上面提到的第二個問題)。

此時可以在訊息源流入Flink流處理系統時，自動生成增量的時間戳賦予訊息，之後處理的流程與Event Time相同。Ingress Time可以看成是Event Time的一個特例，由於其在訊息源處時間戳一定是有序的，

所以在流處理系統中，相對於Event Time，其亂序的訊息延遲不會很高，因此對Flink分散式系統的吞吐量和延遲的影響也會更小。

Event Time時間視窗的實現

Flink借鑑了Google的MillWheel專案，通過WaterMark來支援基於Event Time時間視窗。

當操作符通過基於Event Time的時間視窗來處理資料時，它必須在確定所有屬於該時間視窗的訊息全部流入此操作符後，才能開始處理資料。

但是由於訊息可能是亂序的，所以操作符無法直接確認何時所有屬於該時間視窗的訊息全部流入此操作符。

WaterMark包含一個時間戳，Flink使用WaterMark標記所有小於該時間戳的訊息都已流入，Flink的資料來源在確認所有小於某個時間戳的訊息都已輸出到Flink流處理系統後，

會生成一個包含該時間戳的WaterMark，插入到訊息流中輸出到Flink流處理系統中，Flink操作符按照時間視窗快取所有流入的訊息，當操作符處理到WaterMark時，

它對所有小於該WaterMark時間戳的時間視窗的資料進行處理併發送到下一個操作符節點，然後也將WaterMark傳送到下一個操作符節點。

為了保證能夠處理所有屬於某個時間視窗的訊息，操作符必須等到大於這個時間視窗的WaterMark之後，才能開始對該時間視窗的訊息進行處理，相對於基於Operator Time的時間視窗，

Flink需要佔用更多的記憶體，且會直接影響訊息處理的延遲時間。對此，一個可能的優化措施是，對於聚合類的操作符，可能可以提前對部分訊息進行聚合操作，

當有屬於該時間視窗的新訊息流入時，基於之前的部分聚合結果繼續計算，這樣的話，只需快取中間計算結果即可，無需快取該時間視窗的所有訊息。

對於基於Event Time時間視窗的操作符來說，流入WaterMark的時間戳與當前節點的時鐘一致是最簡單理想的狀況了，但是在實際環境中是不可能的，

由於訊息的亂序以及前面節點處理效率的不同，總是會有某些訊息流入時間大於其本身的時間戳，真實WaterMark時間戳與理想情況下WaterMark時間戳的差別稱為Time Skew，如下圖所示：

圖5 WaterMark的Time Skew圖

Time Skew決定了該WaterMark與上一個WaterMark之間的時間視窗所有資料需要快取的時間，Time Skew時間越長，該時間視窗資料的延遲越長，佔用記憶體的時間也越長，同時會對流處理系統的吞吐量產生負面影響。