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時間越長，該時間窗口數據的延遲越長，占用內存的時間也越長，同時會對流處理系統的吞吐量產生負面影響。

Flink流處理的時間窗口