一、監控系統概覽

監控系統在現代技術環境中扮演著至關重要的角色。運營同學每天檢查自己的活動數據，研發人員每天檢查系統各項指標是否正常，這些工作都少不了監控系統的身影。通常來講，監控系統包括數據采集、數據計算、數據存儲、數據可視化及監控預警等功能。本文主要介紹數據計算部分。

二、實時計算

流數據實時計算是一種處理和分析實時數據流的技術，它允許企業從連續生成的數據（如日志文件、傳感器數據、在線交易等）中即時提取價值。這種計算模式對于需要快速決策和響應的應用場景至關重要，如實時監控、在線推薦、欺詐檢測等。Apache Flink 是實現流數據實時計算的流行框架之一。

?

2.1 數據流

數據流（Data Stream）是由連續生成的數據元素組成的序列，這些數據元素可以是來自各種源的記錄、事件、或者其他形式的數據點。數據流通常是動態的、無界的，并且高速連續地到達處理系統。

數據流的特點包括：

1.連續性：數據流是連續到達的，沒有明確的開始和結束。

2.無界性：理論上，數據流可以無限地持續下去，因此通常被認為是無界的。

3.實時性：數據流通常需要實時或近實時處理，以便及時響應或提取信息。

4.變化性：數據流的特性（如速度、大小、格式）可能會隨時間變化。

5.無序性：數據流中的數據可能不按照產生的順序到達，尤其是在分布式系統中，可能因為網絡延遲或其他因素導致亂序。

?

[數據源] → |元素1| → |元素2| → |元素3| → ... → [數據處理] → [數據存儲/輸出]

?

2.2 數據流處理

2.2.1 流處理中的Time和Window

Time

事件時間（Event Time）

事件時間是指每個事件或元素在其生產設備上產生的時間。該時間通常在它們進入Flink之前就已經嵌入在事件中，并且可以從每個事件中提取事件時間戳。

有了事件時間，基于窗口的聚合（例如，每分鐘的事件數量）只是事件時間列上的一種特殊的分組和聚合。每個時間窗口是一個組，每一行數據可以屬于多個窗口／組（針對滑動窗口，多個窗口可能有重合的數據）。

處理時間（Processing Time）：

處理時間是指正在執行相應Flink操作的機器的系統時間。

當流式程序按處理時間運行時，所有基于時間的操作（如時間窗口）都將使用運行相應操作的計算機的系統時間。在分布式和異步環境中，處理時間不能提供確定性，因為它容易受到記錄到達系統（例如從消息隊列）的速度以及記錄在系統內部操作之間流動的速度的影響。

Window

無界數據流本身沒有邊界，但是對數據流的計算需要一個明確的邊界。這就要將無界數據流劃分為有界數據流，邊界的劃分一般有兩種方式：時間驅動或者數據驅動，時間驅動就是每隔多長時間就劃分一個邊界，數據驅動就是每來多少條數據劃分一個邊界。

?

?

2.2.2 窗口的分類

1. 滾動窗口（Tumbling Window）

滾動窗口將數據流分割成不重疊、連續的時間間隔。每個窗口都是獨立的，窗口長度是固定的。例如，如果設置了一個5分鐘的固定時間窗口，那么數據流會被劃分為0-5分鐘、5-10分鐘、10-15分鐘等時間段。每個窗口都會獨立處理，適用于需要定期重置計數或計算的場景。

?

?

2. 滑動時間窗口（Sliding Window）

滑動時間窗口可以有重疊，它由兩個參數定義：窗口長度和滑動間隔。窗口長度決定了數據聚合的時間范圍，而滑動間隔決定了窗口更新的頻率。例如，如果窗口長度是10分鐘，滑動間隔是5分鐘，那么第一個窗口是0-10分鐘，第二個窗口是5-15分鐘，依此類推。滑動時間窗口適用于需要更平滑連續輸出的場景。

?

?

3. 會話窗口（Session Window）

會話窗口是動態長度的窗口，它根據數據流的活動來定義。會話窗口在數據活動（即事件）發生時開啟，在一定的不活動時間段（稱為超時時間或間隔）之后關閉。這種類型的窗口適用于活動或會話驅動的場景，比如用戶的網頁瀏覽行為分析。

?

4. 全局窗口（Global Window）

全局窗口是一個無限期的窗口，它不會根據時間進行分割。在全局窗口中，數據流的處理通常由其他機制觸發，如外部信號或數據數量達到一定閾值。它不常用，因為大多數流處理場景都需要某種形式的時間邊界來限制數據處理。

2.2.3 窗口的生命周期

窗口創建

窗口不會預先創建好，而是由數據驅動創建。當第一個應該屬于這個窗口的數據元素到達時，就會創建對應的窗口。

窗口計算

對于不同的窗口類型，觸發計算的條件也會不同。例如，一個滾動事件時間窗口，應該在水位線到達窗口結束時間的時候觸發計算；而一個計數窗口，會在窗口中元素數量達到定義大小時觸發計算。

窗口銷毀

一般情況下，當時間達到了結束點，就會直接觸發計算、輸出結果，進而清除狀態、銷毀窗口。這時窗口的銷毀可以認為和觸發計算是同一時刻。這里需要注意，Flink 中只對時間窗口（TimeWindow）有銷毀機制；由于計數窗口（CountWindow）是基于全局窗口（GlobalWindow）實現的，而全局窗口不會清除狀態，所以就不會被銷毀。

?

2.2.4 基于窗口機制的流計算

算子模型

Flink中算子分為數據源算子（source）、轉換算子（transform）、輸出算子（sink），下圖為算子模型示意圖。數據源算子負責接收運算數據，數據源支持多種：文本、MQ等等；轉換算子主要對數據流進行聚合和計算操作；sink算子主要負責將運算結果輸出，包括持久化和轉發運算結果（MQ）等。

?

?

下圖展示了用窗口大小為10s的滾動時間窗口處理數據的例子，數據流中所有數據都按序到達，這是最理想的情況。

?

水位線機制

對于分布式系統而言，各個系統節點相互獨立，互不影響，這給系統帶來了更高的穩定性。但是各個節點之間沒有統一的時鐘，而是各自維護一個邏輯時鐘。數據流在不同節點之間流動，上游節點給下游節點傳輸數據時，不同的下游節點對于時間的處理也會有偏差。如果要統一各個節點之間的時鐘，則需要上游節點給下游節點傳遞數據時，將事件時間也傳遞下來。

以下圖為例，時間戳為12和13的數據分別進如source1和source2算子。source1算子將基于事件時間的邏輯時鐘傳遞給下游operator1算子，operator1算子將本地邏輯時鐘置為12；source2算子將邏輯時鐘傳遞給下游operator2算子，operator2算子將本地邏輯時鐘置為13。這就造成了不同下游節點之間邏輯時鐘不統一的問題。

想要解決這個問題，需要上游算子將邏輯時鐘以廣播形式傳遞出去，并且該邏輯時鐘的傳遞不會受到當前算子作業進度的影響。

水位線可以看做是一種特殊的數據記錄，該記錄中包含了邏輯時鐘，其主要內容就是一個時間戳，并且只能遞增。其表示該時間戳之前的數據都已經到達，結束時間小于該時間戳的窗口都可以觸發計算和關閉窗口。

水位線傳遞機制

有序流的水位線傳遞比較簡單，數據和水位線全部按照自身順序進行傳遞，下游依次處理，當水位線到達了某個算子任務，這個任務會將內部時鐘設置為當前時間戳。

數據流并不總是有序，由于網絡延遲等原因可能會造成數據流亂序。水位線周期性生成時，以當前周期內的最大事件時間進行計算。

一旦算子任務開啟了并行，水位線的傳遞就會變得復雜。以下圖為例，并行任務間的水位線傳遞。

1）operator1算子初始化內部邏輯時鐘為0，其接收兩個并行上游算子source1、source2的結果。

2）source1算子處理事件時間為1的數據，并將水位線1傳遞給下游operator1算子，operator1算子接收到source1的水位線之后，不更新自己的邏輯時鐘。需要等待source2算子發送水位線，并進行比較后才能更新自己的邏輯時鐘。

3）source2算子處理事件時間為3的數據，并將水位線3傳遞給下游operator1算子，operator1算子接收到source2的水位線之后，和source1分區的水位線1進行比較，取最小值[1]作為自己的邏輯時鐘時間。

4）source1算子處理事件時間為4的數據，并將水位線4傳遞給下游operator1算子，operator1算子接收到source1的水位線之后，和source2分區的水位線[4]進行比較，取最小值[3]作為自己的邏輯時鐘時間。

水位線最長等待時間

也就是說，下游算子依賴所有上游算子的水位線來設置自己的邏輯時鐘。假如有一個上游算子一直沒有發送水位線，下游算子的邏輯時鐘則無法更新，這時下游算子無法正常執行自己的計算任務。

此時，需要設置水位線最長等待時間，超過最長等待時間還是沒有接收到某個上游算子的水位線信息，則排除該上游算子，即該上游算子的水位線不再參與比較。這樣下游算子的邏輯時鐘就能夠正常推進。

遲到數據處理

對于數據流而言，數據并不總是按序到達。如果某些數據因為網絡原因導致亂序甚至延遲，這些數據就有無法正確計算的風險。

下圖展示了一個窗口大小為10s的滾動窗口處理數據流的過程：

1）2-6這些數據進入到[0,10)窗口內，此時觸發水位線計算，水位線為w(7)，沒有觸發窗口操作。

2）5-9這兩個數據進入到[0,10)窗口內，11進入到[10,20)窗口內，此時出發水位線計算，水位線為w(11)，當前水位線大于[0,10)窗口的結束時間，觸發該窗口的計算和關閉操作。

3）11進入到[10,20)窗口內后，又有一條數據8來到。此時屬于它的窗口已經觸發計算并關閉，不處理該條數據。

4）15-16這些數據進入到[10,20)窗口內，此時觸發水位線計算，水位線為w(16)，沒有觸發窗口操作。

5）17-20這些數據進入到[10,20)窗口內，此時觸發水位線計算，水位線為w(20)。當前水位線大于[10,20)窗口的結束時間，觸發該窗口的計算和關閉操作。

?

為了解決亂序流中延遲數據的處理問題，提出了水位線延遲時間的概念。例如，想實現水位線延遲兩秒，則只需要將當前事件時間減去兩秒作為當前的水位線即可。以上圖為例，時間窗口大小為10s，將水位線延遲時間設置為兩秒，再次進行數據流處理。

1）2-6這些數據進入到[0,10)窗口內，此時觸發水位線計算，水位線為w(5)，沒有觸發窗口操作。

2）5-9這兩個數據進入到[0,10)窗口內，11進入到[10,20)窗口內，此時出發水位線計算，水位線為w(9)，沒有觸發窗口操作。

3）11進入到[10,20)窗口內后，又有一條數據8來到，該條數據進入到[0,10)窗口內。

4）15-16這些數據進入到[10,20)窗口內，此時觸發水位線計算，水位線為w(12)，當前水位線大于[0,10)窗口的結束時間，觸發該窗口的計算和關閉操作。

5）17-20這些數據進入到[10,20)窗口內，此時觸發水位線計算，水位線為w(18)，不觸發窗口操作。其中亂序數據4無法處理。

?

通過設置水位線延遲等待時間，可以處理一些輕微延遲的數據。如果數據延遲非常嚴重，在水位線等待時間內還是沒有等到對應窗口的數據，又該怎么辦呢？

其實水位線觸發窗口計算和關閉是兩個動作，觸發窗口計算之后如果窗口不進行關閉，那么延遲嚴重的數據還可以通過側輸出流進入到該窗口再次出發計算。Flink也支持設置窗口關閉延遲時間，將窗口關閉延遲時間設置為5s，以上圖為例，再次進行數據流處理。

1）2-6這些數據進入到[0,10)窗口內，此時觸發水位線計算，水位線為w(5)，沒有觸發窗口操作。

2）5-9這兩個數據進入到[0,10)窗口內，11進入到[10,20)窗口內，此時出發水位線計算，水位線為w(9)，沒有觸發窗口操作。

3）11進入到[10,20)窗口內后，又有一條數據8來到，該條數據進入到[0,10)窗口內。

4）15-16這些數據進入到[10,20)窗口內，此時觸發水位線計算，水位線為w(12)，當前水位線大于[0,10)窗口的結束時間，觸發該窗口的計算但不關閉該窗口。

5）數據17進入到[10,20)窗口內，此時又來了數據4，由于當前已經超過了水位線延遲時間，數據無法直接進入[0,10)窗口內。通過側輸出流進入到窗口內再次觸發窗口計算。

6）19-20進入到[10,20)窗口內，此時出發水位線計算，水位線為w(18)，當前水位線大于[0,10)窗口的延遲關閉時間，關閉該窗口。

通過設置窗口延遲計算和延遲關閉在一定程度上可以解決數據遲到的問題，在實時計算場景，窗口計算延遲設置不宜過大，否則會失去結果的實時性。

?

寫在最后

由于篇幅有限，這篇文章只介紹了數據計算中的一部分，要想實現容錯性高、計算精準的數據計算服務，需要考慮很多場景。例如，算子并行計算時，某個算子突然不可用，如何恢復數據的計算結果？并行算子間，每個算子的計算負載不一致又該如何處理？歡迎大家一起討論~

審核編輯 黃宇