1.4 根據事件時間開滾動窗口

假定某個無界數據集在事件時間區間[12：00，12：08）內有10條記錄，每條記錄的值都是整數。以事件時間為橫軸，以處理時間（觀察時間）為縱軸，記錄以圓點表示，則所有記錄在空間中的位置如圖1-24所示。為了便于后續引用，這個例子被命名為Example1.1。

其中，由于傳輸通道延遲，處理時間落后于事件時間 5分鐘，即縱坐標軸的零點代表處理時間12：05。

根據事件時間開滾動窗口，窗口大小為 2分鐘，本節的任務是計算每個窗口內記錄值的和。

1.4.1 what：轉換/where：窗口

Example1.1中的窗口為事件時間窗口，分別為[12：00，12：02）、[12：02，12：04）、[12：04，12：06）和[12：06，12：08）。轉換操作為求和（聚合），聚合結果如圖 1-25所示。

聚合操作發生在什么處理時間點上呢？圖1-25中假定所有聚合操作發生在處理時間點12：10，水印是解決這個問題的方案。

1.4.2 when：水印

由于事先并不知道橫軸的每個窗格里有多少條記錄，我們并不能確定何時觸發聚合操作，水印則能標記出這個時間點。

水印能夠標記出某個事件時間點以前的所有記錄均已到達引擎。例如，在窗口[12：00，12：02）內觀察到記錄9的同時也觀察到代表水印的時間戳12：02，則我們可以在記錄9被觀察到的時間點上正確地觸發聚合操作。

水印可抽象地表示成函數f（P）=E，其中F代表處理時間，E代表事件時間，即我們能夠在處理時間點P判定事件時間推進到了E。

圖1-26描繪了水印、事件時間和處理時間的關系。

考慮到系統開銷，水印是離散的，即只有部分記錄后附有水印。為了便于分析，水印通常以連續曲線繪制。此外，作為事件時間的推進器，水印曲線是單調遞增的。

有以下兩類水印。

（1）完美水印（Perfect Watermark）：完美水印表示早于水印標記事件時間戳的所有記錄均已到達，非亂序的無界數據集中最近一條記錄的事件時間就是完美水印。

（2）啟發式水?。℉euristic Watermark）：啟發式水印是盡可能地確定時間戳的一種估計，可能出現某些事件晚于水印到達的情況。在分布式系統中，定義完美水印往往是非常困難的，定義啟發式水印的代價則相對較低。

分別嵌入完美水印和啟發式水印時窗口的聚合情況，如圖1-27所示。

可以看出，在嵌入啟發式水印時，記錄 9由于水印遲到而沒有計入對應事件時間窗口的聚合結果內。

在解決這個問題時，這兩類水印都存在缺陷。

（1）水印遲到：在嵌入完美水印時，由于記錄9在處理時間軸上推進得太慢，事件時間窗口[12：02，12：04）和[12：04，12：06）的聚合操作被推遲到處理時間點12：08之后，這與低延遲計算的目標相悖；同時，會拉長這兩個窗口的生存期，即這兩個窗口所占用的資源不能及時釋放。

（2）水印早到：在嵌入啟發式水印時，事件時間窗口[12：00，12：02）在處理時間軸上推進得太快，導致記錄 9沒能計入本窗口的聚合結果內，這與精準計算的目標相悖，引擎應提供事后更正機制。

1.4.3 when：觸發器

作為定義轉換操作時間點的另一類方案，觸發器解決了水印有缺陷的問題。

（1）在嵌入完美水印時，事件時間窗口[12：02，12：04）的聚合操作被推遲到處理時間點 12：08 之后，這與低延遲計算的目標相悖。為此，類比將事件時間軸劃分為長度為 2 分鐘的窗格（pane）而得到事件時間窗口，我們將處理時間軸劃分為長度為1分鐘的窗格，如[12：05，12：06）、[12：06，12：07）、[12：07，12：08）、[12：08，12：09）、[12：09，12：10），然后在每個窗格邊界處觸發一次聚合計算，這樣先后得到實時（圖 1-28（a）中以“早到”標注）聚合結果 7、14 和 22。由于在得到聚合結果22時水印還沒有被觀察到，這個窗口仍需保留至處理時間點12：09。

（2）在嵌入完美水印時，事件時間窗口[12：04，12：06）的聚合操作也被推遲到處理時間點 12：08 之后，按照上述方案需要分別在處理時間窗格[12：05，12：06）、[12：06，12：07）、[12：07，12：08）、[12：08，12：09）內觸發聚合計算。由于這個事件時間窗口內只有處理時間窗格[12：06，12：07）內有記錄，因此在其他窗格內觸發聚合計算沒有意義，可以通過定義事件數量觸發器解決這個問題。在本例中可以定義事件數量為1。

（3）在嵌入啟發式水印時，在事件時間窗口[12：02，12：04）的處理時間窗格[12：07，12：08）內觀察到水印，因此我們不是在處理時間窗格[12：07，12：08）的邊界處觸發聚合計算，而是按照水印的推進時間觸發聚合計算，這會按時（on-time）聚合出結果22。

（4）在嵌入啟發式水印的事件時間窗口[12：00，12：02）內，水印跟隨記錄5到達，我們會按時得到聚合結果 5。為了得到精準的聚合結果，我們必須延長這個時間窗口的生存期，但是由于并不能確切地獲悉還有多少遲到的記錄，如何確定這個時間窗口的生存期是個問題。

因此，可以根據水印、處理時間軸窗格和事件數量確定在處理時間軸的什么地方觸發聚合計算，如圖1-28所示。

截至目前，這個數據處理設計還存在兩個需要解決的問題。

（1）在啟發式水印早到時，為了確保精準計算，引擎必須延長對應事件時間窗口的生存期，這會加大引擎的內存消耗。

（2）同一個事件時間窗口的多個處理時間窗格會輸出多個聚合結果，引擎需要提供定義這些結果之間關系的機制。

可以利用遲到生存期（Allowed Lateness）解決第一個問題，利用累加（Accumulation）模式解決第二個問題。

1.4.4 when：遲到生存期

在嵌入完美水印時，事件不會遲到，窗口能夠及時銷毀；在啟發式水印早到時，為了確保精準計算，引擎必須延長對應事件時間窗口的生存期，所以遲到生存期只會發生在嵌入啟發式水印時。

假定遲到生存期為1分鐘，下面以事件窗口[12：00，12：02）為例進行分析。

（1）根據水印函數曲線計算出本窗口生存期結束的事件時間點所對應的處理時間點。本窗口生存期結束于事件時間 12：03（12：02+0：01），從水印曲線上找到這個事件時間點對應于處理時間軸上的12：07～12：08之間的某個時間點，記為A。

這里需要再次強調生存期是事件時間，但是決定銷毀窗口的時間點是處理時間。在圖1-24中，我們均勻地標注了事件時間坐標點，如12：01、12：02等，但實際上事件時間的推進并不是均勻的，所以我們不能通過處理時間的推進（間隔）推斷事件時間的推進，這也是為什么要從水印曲線上找到窗口的處理時間結束點（A）的原因。

（2）如果在處理時間結束點之前觀察到事件，則應再次觸發聚合計算；在處理時間結束點之后，本窗口被銷毀。因此，可以看出遲到生存期和水印一樣都是聚合計算的觸發信號?；谶@種定義，記錄9不會被丟棄。

我們設定遲到生存期為1分鐘的聚合情況，如圖1-29所示。

在圖1-29中，平行于事件時間軸的虛線標記本窗口的處理時間結束點。

1.4.5 how：累加模式

處理時間軸窗格會多次觸發聚合計算，累加模式定義同一個事件時間窗口的多個聚合結果之間的關系。有以下三種累加模式。

● 丟棄（discarding）：啟發式水印的事件時間窗口[12：02，12：04）將產生三個聚合結果，在丟棄的模式下分別為7、7和8，即每個窗格內的聚合結果和其他窗格無關。

● 累加（accumulating）：每個窗格會累加前一個相鄰窗格的聚合結果。

● 撤回（retracting）：這種模式是在累加模式的基礎上增加一個撤回結果。啟發式水印的事件時間窗口[12：02，12：04）的第二個窗格聚合結果為-7和14，-7代表撤回，14代表截至當前處理時間總的聚合結果。

這三種模式的聚合情況，分別如圖1-30～圖1-32所示。