監(jiān)控系統(tǒng)原理揭秘-數(shù)據(jù)運(yùn)算篇

一、監(jiān)控系統(tǒng)概覽

監(jiān)控系統(tǒng)在現(xiàn)代技術(shù)環(huán)境中扮演著至關(guān)重要的角色。運(yùn)營同學(xué)每天檢查自己的活動(dòng)數(shù)據(jù)，研發(fā)人員每天檢查系統(tǒng)各項(xiàng)指標(biāo)是否正常，這些工作都少不了監(jiān)控系統(tǒng)的身影。通常來講，監(jiān)控系統(tǒng)包括數(shù)據(jù)采集、數(shù)據(jù)計(jì)算、數(shù)據(jù)存儲(chǔ)、數(shù)據(jù)可視化及監(jiān)控預(yù)警等功能。本文主要介紹數(shù)據(jù)計(jì)算部分。

二、實(shí)時(shí)計(jì)算

流數(shù)據(jù)實(shí)時(shí)計(jì)算是一種處理和分析實(shí)時(shí)數(shù)據(jù)流的技術(shù)，它允許企業(yè)從連續(xù)生成的數(shù)據(jù)（如日志文件、傳感器數(shù)據(jù)、在線交易等）中即時(shí)提取價(jià)值。這種計(jì)算模式對于需要快速?zèng)Q策和響應(yīng)的應(yīng)用場景至關(guān)重要，如實(shí)時(shí)監(jiān)控、在線推薦、欺詐檢測等。Apache Flink 是實(shí)現(xiàn)流數(shù)據(jù)實(shí)時(shí)計(jì)算的流行框架之一。

?

2.1 數(shù)據(jù)流

數(shù)據(jù)流（Data Stream）是由連續(xù)生成的數(shù)據(jù)元素組成的序列，這些數(shù)據(jù)元素可以是來自各種源的記錄、事件、或者其他形式的數(shù)據(jù)點(diǎn)。數(shù)據(jù)流通常是動(dòng)態(tài)的、無界的，并且高速連續(xù)地到達(dá)處理系統(tǒng)。

數(shù)據(jù)流的特點(diǎn)包括：

1.連續(xù)性：數(shù)據(jù)流是連續(xù)到達(dá)的，沒有明確的開始和結(jié)束。

2.無界性：理論上，數(shù)據(jù)流可以無限地持續(xù)下去，因此通常被認(rèn)為是無界的。

3.實(shí)時(shí)性：數(shù)據(jù)流通常需要實(shí)時(shí)或近實(shí)時(shí)處理，以便及時(shí)響應(yīng)或提取信息。

4.變化性：數(shù)據(jù)流的特性（如速度、大小、格式）可能會(huì)隨時(shí)間變化。

5.無序性：數(shù)據(jù)流中的數(shù)據(jù)可能不按照產(chǎn)生的順序到達(dá)，尤其是在分布式系統(tǒng)中，可能因?yàn)?a href="http://ttokpm.com/v/tag/1722/" target="_blank">網(wǎng)絡(luò)延遲或其他因素導(dǎo)致亂序。

?

[數(shù)據(jù)源] → |元素1| → |元素2| → |元素3| → ... → [數(shù)據(jù)處理] → [數(shù)據(jù)存儲(chǔ)/輸出]

?

2.2 數(shù)據(jù)流處理

2.2.1 流處理中的Time和Window

Time

事件時(shí)間（Event Time）

事件時(shí)間是指每個(gè)事件或元素在其生產(chǎn)設(shè)備上產(chǎn)生的時(shí)間。該時(shí)間通常在它們進(jìn)入Flink之前就已經(jīng)嵌入在事件中，并且可以從每個(gè)事件中提取事件時(shí)間戳。

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

處理時(shí)間（Processing Time）：

處理時(shí)間是指正在執(zhí)行相應(yīng)Flink操作的機(jī)器的系統(tǒng)時(shí)間。

當(dāng)流式程序按處理時(shí)間運(yùn)行時(shí)，所有基于時(shí)間的操作（如時(shí)間窗口）都將使用運(yùn)行相應(yīng)操作的計(jì)算機(jī)的系統(tǒng)時(shí)間。在分布式和異步環(huán)境中，處理時(shí)間不能提供確定性，因?yàn)樗菀资艿接涗浀竭_(dá)系統(tǒng)（例如從消息隊(duì)列）的速度以及記錄在系統(tǒng)內(nèi)部操作之間流動(dòng)的速度的影響。

Window

無界數(shù)據(jù)流本身沒有邊界，但是對數(shù)據(jù)流的計(jì)算需要一個(gè)明確的邊界。這就要將無界數(shù)據(jù)流劃分為有界數(shù)據(jù)流，邊界的劃分一般有兩種方式：時(shí)間驅(qū)動(dòng)或者數(shù)據(jù)驅(qū)動(dòng)，時(shí)間驅(qū)動(dòng)就是每隔多長時(shí)間就劃分一個(gè)邊界，數(shù)據(jù)驅(qū)動(dòng)就是每來多少條數(shù)據(jù)劃分一個(gè)邊界。

?

?

2.2.2 窗口的分類

1. 滾動(dòng)窗口（Tumbling Window）

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

?

?

2. 滑動(dòng)時(shí)間窗口（Sliding Window）

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

?

?

3. 會(huì)話窗口（Session Window）

會(huì)話窗口是動(dòng)態(tài)長度的窗口，它根據(jù)數(shù)據(jù)流的活動(dòng)來定義。會(huì)話窗口在數(shù)據(jù)活動(dòng)（即事件）發(fā)生時(shí)開啟，在一定的不活動(dòng)時(shí)間段（稱為超時(shí)時(shí)間或間隔）之后關(guān)閉。這種類型的窗口適用于活動(dòng)或會(huì)話驅(qū)動(dòng)的場景，比如用戶的網(wǎng)頁瀏覽行為分析。

?

4. 全局窗口（Global Window）

全局窗口是一個(gè)無限期的窗口，它不會(huì)根據(jù)時(shí)間進(jìn)行分割。在全局窗口中，數(shù)據(jù)流的處理通常由其他機(jī)制觸發(fā)，如外部信號(hào)或數(shù)據(jù)數(shù)量達(dá)到一定閾值。它不常用，因?yàn)榇蠖鄶?shù)流處理場景都需要某種形式的時(shí)間邊界來限制數(shù)據(jù)處理。

2.2.3 窗口的生命周期

窗口創(chuàng)建

窗口不會(huì)預(yù)先創(chuàng)建好，而是由數(shù)據(jù)驅(qū)動(dòng)創(chuàng)建。當(dāng)?shù)谝粋€(gè)應(yīng)該屬于這個(gè)窗口的數(shù)據(jù)元素到達(dá)時(shí)，就會(huì)創(chuàng)建對應(yīng)的窗口。

窗口計(jì)算

對于不同的窗口類型，觸發(fā)計(jì)算的條件也會(huì)不同。例如，一個(gè)滾動(dòng)事件時(shí)間窗口，應(yīng)該在水位線到達(dá)窗口結(jié)束時(shí)間的時(shí)候觸發(fā)計(jì)算；而一個(gè)計(jì)數(shù)窗口，會(huì)在窗口中元素?cái)?shù)量達(dá)到定義大小時(shí)觸發(fā)計(jì)算。

窗口銷毀

一般情況下，當(dāng)時(shí)間達(dá)到了結(jié)束點(diǎn)，就會(huì)直接觸發(fā)計(jì)算、輸出結(jié)果，進(jìn)而清除狀態(tài)、銷毀窗口。這時(shí)窗口的銷毀可以認(rèn)為和觸發(fā)計(jì)算是同一時(shí)刻。這里需要注意，F(xiàn)link 中只對時(shí)間窗口（TimeWindow）有銷毀機(jī)制；由于計(jì)數(shù)窗口（CountWindow）是基于全局窗口（GlobalWindow）實(shí)現(xiàn)的，而全局窗口不會(huì)清除狀態(tài)，所以就不會(huì)被銷毀。

?

2.2.4 基于窗口機(jī)制的流計(jì)算

算子模型

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

?

?

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

?

水位線機(jī)制

對于分布式系統(tǒng)而言，各個(gè)系統(tǒng)節(jié)點(diǎn)相互獨(dú)立，互不影響，這給系統(tǒng)帶來了更高的穩(wěn)定性。但是各個(gè)節(jié)點(diǎn)之間沒有統(tǒng)一的時(shí)鐘，而是各自維護(hù)一個(gè)邏輯時(shí)鐘。數(shù)據(jù)流在不同節(jié)點(diǎn)之間流動(dòng)，上游節(jié)點(diǎn)給下游節(jié)點(diǎn)傳輸數(shù)據(jù)時(shí)，不同的下游節(jié)點(diǎn)對于時(shí)間的處理也會(huì)有偏差。如果要統(tǒng)一各個(gè)節(jié)點(diǎn)之間的時(shí)鐘，則需要上游節(jié)點(diǎn)給下游節(jié)點(diǎn)傳遞數(shù)據(jù)時(shí)，將事件時(shí)間也傳遞下來。

以下圖為例，時(shí)間戳為12和13的數(shù)據(jù)分別進(jìn)如source1和source2算子。source1算子將基于事件時(shí)間的邏輯時(shí)鐘傳遞給下游operator1算子，operator1算子將本地邏輯時(shí)鐘置為12；source2算子將邏輯時(shí)鐘傳遞給下游operator2算子，operator2算子將本地邏輯時(shí)鐘置為13。這就造成了不同下游節(jié)點(diǎn)之間邏輯時(shí)鐘不統(tǒng)一的問題。

想要解決這個(gè)問題，需要上游算子將邏輯時(shí)鐘以廣播形式傳遞出去，并且該邏輯時(shí)鐘的傳遞不會(huì)受到當(dāng)前算子作業(yè)進(jìn)度的影響。

水位線可以看做是一種特殊的數(shù)據(jù)記錄，該記錄中包含了邏輯時(shí)鐘，其主要內(nèi)容就是一個(gè)時(shí)間戳，并且只能遞增。其表示該時(shí)間戳之前的數(shù)據(jù)都已經(jīng)到達(dá)，結(jié)束時(shí)間小于該時(shí)間戳的窗口都可以觸發(fā)計(jì)算和關(guān)閉窗口。

水位線傳遞機(jī)制

有序流的水位線傳遞比較簡單，數(shù)據(jù)和水位線全部按照自身順序進(jìn)行傳遞，下游依次處理，當(dāng)水位線到達(dá)了某個(gè)算子任務(wù)，這個(gè)任務(wù)會(huì)將內(nèi)部時(shí)鐘設(shè)置為當(dāng)前時(shí)間戳。

數(shù)據(jù)流并不總是有序，由于網(wǎng)絡(luò)延遲等原因可能會(huì)造成數(shù)據(jù)流亂序。水位線周期性生成時(shí)，以當(dāng)前周期內(nèi)的最大事件時(shí)間進(jìn)行計(jì)算。

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

1）operator1算子初始化內(nèi)部邏輯時(shí)鐘為0，其接收兩個(gè)并行上游算子source1、source2的結(jié)果。

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

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

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

水位線最長等待時(shí)間

也就是說，下游算子依賴所有上游算子的水位線來設(shè)置自己的邏輯時(shí)鐘。假如有一個(gè)上游算子一直沒有發(fā)送水位線，下游算子的邏輯時(shí)鐘則無法更新，這時(shí)下游算子無法正常執(zhí)行自己的計(jì)算任務(wù)。

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

遲到數(shù)據(jù)處理

對于數(shù)據(jù)流而言，數(shù)據(jù)并不總是按序到達(dá)。如果某些數(shù)據(jù)因?yàn)榫W(wǎng)絡(luò)原因?qū)е聛y序甚至延遲，這些數(shù)據(jù)就有無法正確計(jì)算的風(fēng)險(xiǎn)。

下圖展示了一個(gè)窗口大小為10s的滾動(dòng)窗口處理數(shù)據(jù)流的過程：

1）2-6這些數(shù)據(jù)進(jìn)入到[0,10)窗口內(nèi)，此時(shí)觸發(fā)水位線計(jì)算，水位線為w(7)，沒有觸發(fā)窗口操作。

2）5-9這兩個(gè)數(shù)據(jù)進(jìn)入到[0,10)窗口內(nèi)，11進(jìn)入到[10,20)窗口內(nèi)，此時(shí)出發(fā)水位線計(jì)算，水位線為w(11)，當(dāng)前水位線大于[0,10)窗口的結(jié)束時(shí)間，觸發(fā)該窗口的計(jì)算和關(guān)閉操作。

3）11進(jìn)入到[10,20)窗口內(nèi)后，又有一條數(shù)據(jù)8來到。此時(shí)屬于它的窗口已經(jīng)觸發(fā)計(jì)算并關(guān)閉，不處理該條數(shù)據(jù)。

4）15-16這些數(shù)據(jù)進(jìn)入到[10,20)窗口內(nèi)，此時(shí)觸發(fā)水位線計(jì)算，水位線為w(16)，沒有觸發(fā)窗口操作。

5）17-20這些數(shù)據(jù)進(jìn)入到[10,20)窗口內(nèi)，此時(shí)觸發(fā)水位線計(jì)算，水位線為w(20)。當(dāng)前水位線大于[10,20)窗口的結(jié)束時(shí)間，觸發(fā)該窗口的計(jì)算和關(guān)閉操作。

?

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

1）2-6這些數(shù)據(jù)進(jìn)入到[0,10)窗口內(nèi)，此時(shí)觸發(fā)水位線計(jì)算，水位線為w(5)，沒有觸發(fā)窗口操作。

2）5-9這兩個(gè)數(shù)據(jù)進(jìn)入到[0,10)窗口內(nèi)，11進(jìn)入到[10,20)窗口內(nèi)，此時(shí)出發(fā)水位線計(jì)算，水位線為w(9)，沒有觸發(fā)窗口操作。

3）11進(jìn)入到[10,20)窗口內(nèi)后，又有一條數(shù)據(jù)8來到，該條數(shù)據(jù)進(jìn)入到[0,10)窗口內(nèi)。

4）15-16這些數(shù)據(jù)進(jìn)入到[10,20)窗口內(nèi)，此時(shí)觸發(fā)水位線計(jì)算，水位線為w(12)，當(dāng)前水位線大于[0,10)窗口的結(jié)束時(shí)間，觸發(fā)該窗口的計(jì)算和關(guān)閉操作。

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

?

通過設(shè)置水位線延遲等待時(shí)間，可以處理一些輕微延遲的數(shù)據(jù)。如果數(shù)據(jù)延遲非常嚴(yán)重，在水位線等待時(shí)間內(nèi)還是沒有等到對應(yīng)窗口的數(shù)據(jù)，又該怎么辦呢？

其實(shí)水位線觸發(fā)窗口計(jì)算和關(guān)閉是兩個(gè)動(dòng)作，觸發(fā)窗口計(jì)算之后如果窗口不進(jìn)行關(guān)閉，那么延遲嚴(yán)重的數(shù)據(jù)還可以通過側(cè)輸出流進(jìn)入到該窗口再次出發(fā)計(jì)算。Flink也支持設(shè)置窗口關(guān)閉延遲時(shí)間，將窗口關(guān)閉延遲時(shí)間設(shè)置為5s，以上圖為例，再次進(jìn)行數(shù)據(jù)流處理。

1）2-6這些數(shù)據(jù)進(jìn)入到[0,10)窗口內(nèi)，此時(shí)觸發(fā)水位線計(jì)算，水位線為w(5)，沒有觸發(fā)窗口操作。

2）5-9這兩個(gè)數(shù)據(jù)進(jìn)入到[0,10)窗口內(nèi)，11進(jìn)入到[10,20)窗口內(nèi)，此時(shí)出發(fā)水位線計(jì)算，水位線為w(9)，沒有觸發(fā)窗口操作。

3）11進(jìn)入到[10,20)窗口內(nèi)后，又有一條數(shù)據(jù)8來到，該條數(shù)據(jù)進(jìn)入到[0,10)窗口內(nèi)。

4）15-16這些數(shù)據(jù)進(jìn)入到[10,20)窗口內(nèi)，此時(shí)觸發(fā)水位線計(jì)算，水位線為w(12)，當(dāng)前水位線大于[0,10)窗口的結(jié)束時(shí)間，觸發(fā)該窗口的計(jì)算但不關(guān)閉該窗口。

5）數(shù)據(jù)17進(jìn)入到[10,20)窗口內(nèi)，此時(shí)又來了數(shù)據(jù)4，由于當(dāng)前已經(jīng)超過了水位線延遲時(shí)間，數(shù)據(jù)無法直接進(jìn)入[0,10)窗口內(nèi)。通過側(cè)輸出流進(jìn)入到窗口內(nèi)再次觸發(fā)窗口計(jì)算。

6）19-20進(jìn)入到[10,20)窗口內(nèi)，此時(shí)出發(fā)水位線計(jì)算，水位線為w(18)，當(dāng)前水位線大于[0,10)窗口的延遲關(guān)閉時(shí)間，關(guān)閉該窗口。

通過設(shè)置窗口延遲計(jì)算和延遲關(guān)閉在一定程度上可以解決數(shù)據(jù)遲到的問題，在實(shí)時(shí)計(jì)算場景，窗口計(jì)算延遲設(shè)置不宜過大，否則會(huì)失去結(jié)果的實(shí)時(shí)性。

?

寫在最后

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

審核編輯 黃宇