文章內(nèi)容由K2研究院原創(chuàng)，轉(zhuǎn)載請注明來源。
本文介紹了流計算的概念和技術(shù)要素，簡單比較了三種主流的流計算框架Structured Streaming（Spark）、Flink和Kafka Streams。第1、2兩節(jié)的目的是希望讀者清晰理解流計算的一些重要概念和技術(shù)要點，尤其是其中一些容易混淆的地方，例如流計算與實時計算的關(guān)系、窗口化及水位線（watermark）和一致性模型等。第3節(jié)比較了一些主流計算框架，旨在為那些需要做系統(tǒng)選型的讀者提供一些參考。
閱讀本文不要求讀者對流計算或其它大數(shù)據(jù)技術(shù)已經(jīng)有一定基礎(chǔ)。但在閱讀第3節(jié)關(guān)于流計算框架比較之前，如果對Hadoop和Spark有一些經(jīng)驗，能更好理解架構(gòu)上的討論。
1
流計算的概念
1.1
流計算是什么？
“流”一般是指源源不斷的數(shù)據(jù)（Unbounded Data）。在理解流計算之前，需要先理解其對立概念——批計算（Batch Computing）。批計算假定在計算發(fā)生之前已經(jīng)獲得待處理的全部數(shù)據(jù)，因此可以一次處理所有的數(shù)據(jù)。相對應(yīng)的，流計算是指發(fā)生在源源不斷的數(shù)據(jù)之上的計算，因此在計算發(fā)生時刻數(shù)據(jù)并未完全抵達，甚至尚未產(chǎn)生。流計算的過程就像工廠里的流水線一樣——一件產(chǎn)品在傳送帶上經(jīng)歷多道工序，其中每一道工序都處理從上游源源不斷輸送來的加工件，處理完成后再向下游輸送。流計算與流水線有許多相似性，例如流水線一般都有多道加工工序，對應(yīng)流計算過程中的多個算子；工廠里一般有多條流水線并行加工，對應(yīng)流計算的并行計算；在流水線的某一道工序上，可能需要等接收到上游一批加工件之后才開始加工，對應(yīng)流計算里的窗口化；等等。
1.2
流計算不等于實時計算
按流計算的定義，流計算只與處理數(shù)據(jù)是否有限（即bounded還是unbounded）有關(guān)。但人們提到流計算，往往會聯(lián)想到一些屬性，比如“實時”、“低延遲”等等。實際上流計算與這些屬性之間并沒有直接聯(lián)系。例如我們每天定時收到天氣預(yù)報，而氣象數(shù)據(jù)顯然是一種流數(shù)據(jù)。天氣預(yù)報系統(tǒng)在每天播報前處理一段時間的氣象數(shù)據(jù)，那么數(shù)據(jù)的處理延遲可達數(shù)小時。
那么為什么人們會產(chǎn)生“實時”或“低延遲”的聯(lián)想呢？這主要受到早期實時計算引擎的影響。實時計算是面向低延遲場景產(chǎn)生的一類計算框架，例如早期的Storm（Storm的新版本也在逐漸豐富流計算的語義，如窗口化）：
Apache Storm is a free and open source distributed real-time computation system.
實時計算與流計算的共同點在于兩者都可以處理流數(shù)據(jù)，但區(qū)別在于兩者的出發(fā)點不同。在成熟的流計算引擎出現(xiàn)之前，人們采用基于分批的模式處理流數(shù)據(jù)（類似下圖），源源不斷的數(shù)據(jù)按照抵達系統(tǒng)時間分批依次處理。
但在一些場景下這種方式的延遲過高，于是人們引入了實時計算來彌補批計算的不足。例如在運營社交網(wǎng)站時，我們可能既希望準確統(tǒng)計前一天的熱點詞匯（這采用批計算），又希望在當天實時發(fā)現(xiàn)一些動態(tài)情況（這采用實時計算）。實時計算最初設(shè)計的出發(fā)點就是降低計算延遲，在語義上無法兼顧批計算的場景，只能與批計算形成一種互補關(guān)系。
1.3
Lambda體系架構(gòu)
基于實時計算與批計算的互補關(guān)系，Nathan Marz提出了一種稱為Lambda的系統(tǒng)架構(gòu)，其本質(zhì)是批計算與實時計算的一種混合架構(gòu)，如下圖所示：
在Lambda體系下，同一份數(shù)據(jù)會分別進入批計算子系統(tǒng)（batch sub-system）和實時計算子系統(tǒng)（realtime sub-system）。批計算子系統(tǒng)存儲完整的數(shù)據(jù)，計算出完整的、精確的結(jié)果；實時計算子系統(tǒng)只處理最近的數(shù)據(jù)，計算出低延遲的、粗糙的結(jié)果。于是，一個Lambda系統(tǒng)兼顧結(jié)果的實時性和準確性。Lambda看起來非常完美，卻存在一些問題。首先，運維兩套子系統(tǒng)是比較繁瑣的；此外，很多時候兩個子系統(tǒng)要完成的數(shù)據(jù)處理邏輯可能是相同的，但卻分別在兩個不同的子系統(tǒng)，存在較高的同步代價。
早期的實時計算引擎在實現(xiàn)上存在一致性問題。例如，大多數(shù)早期的流計算引擎只支持At-most-once（在2.5節(jié)會進一步說明）的處理語義。這意味著一些數(shù)據(jù)可能根本沒有被處理就丟棄了，而另一些數(shù)據(jù)可能被重復(fù)處理了，計算得到的結(jié)果可能根本不正確。
1.4
批流結(jié)合
批計算和流計算之間并不存在本質(zhì)上的差別。批計算是流計算的一種特例，流計算可以處理無限數(shù)據(jù)，當然也能處理有限數(shù)據(jù)。人們之所以接受Lambda這樣的體系是受限于當時缺乏成熟的流計算引擎。隨著人們對流計算的理解不斷深刻和流計算引擎的不斷完善和豐富，出現(xiàn)了一系列能同時兼顧批處理和流處理的計算引擎，如Flink、Structured Streaming、Samza、Kafka Streams等等；還出現(xiàn)了Beam這樣的抽象的計算定義模型。在這些引擎中雖然有時候仍然區(qū)分批計算和流計算的API，但兩者的底層實現(xiàn)往往沒有本質(zhì)區(qū)別，僅為了方便用戶理解。本文后續(xù)討論里提到的流計算引擎不再是在Lambda架構(gòu)中的為了追求低延遲而放棄準確性的實時計算子系統(tǒng)，而特指這些新的支持批流結(jié)合的計算引擎。
2
流計算的技術(shù)要素
在不同的流計算引擎中使用的詞匯可能不一樣，所以為了方便討論，這里借用Beam的詞匯來討論。Beam是一種建立在流計算引擎之上的描述性模型，其抽象層次更高。當然，由于Beam是基于Google Cloud Dataflow提出的，自然帶有其特色，但仍然不妨礙作為理解流計算概念的入口。
從本質(zhì)上看，流計算引擎所做的工作就是：對于源源不斷的數(shù)據(jù)，怎么計算？如何緩存數(shù)據(jù)？在什么時機計算？如何更新計算結(jié)果？不同的流計算引擎的API基本上都是為了描述上面的4個核心問題。在Beam里，把這些問題概括為What、Where、When、How：
What is being computed? 怎么計算？
Where in event time? 要處理發(fā)生在那段時間的數(shù)據(jù)？
When in processing time? 引擎在什么時機計算結(jié)果？
How do refinements relate? 如何更新結(jié)果？
本文沒有討論第一個問題，即“What”。這是因為各種流計算引擎都提供了較為完備的計算算子，大家可以閱讀Flink 、Structured Streaming或Spark Stream關(guān)于算子的描述文檔，而這些內(nèi)容比較直觀，容易理解。一個流計算應(yīng)用往往表示成數(shù)據(jù)源節(jié)點、數(shù)據(jù)輸出節(jié)點和一些算子構(gòu)成的有向圖，如下圖所示：
在下面，我們介紹流計算中一些非常重要的技術(shù)要素。在3.1和3.2中將討論流計算引擎中時間和狀態(tài)兩個重要概念；在3.3里將基于這兩個重要概念，討論算子狀態(tài)在時間上的管理策略（即窗口化），解決上面的“Where”問題；在3.4節(jié)討論計算的時機，解決上面提到的“When”和“How”兩個問題；在3.5中我們將著重討論流計算引擎的一致性定義，這對理解引擎對正確性的保證非常重要。
2.1
發(fā)生時間和處理時間
流計算引擎中的時間一般有兩種情況：
發(fā)生時間（Event time）：指某個事件發(fā)生的時間，通常由外部系統(tǒng)提供；
處理時間（Processing time）：某個事件被處理的時間，在本文里通常指流計算引擎的系統(tǒng)時間。
上述兩個概念比較容易理解。假設(shè)我們正在開發(fā)一套用于設(shè)備高溫報警的系統(tǒng)，設(shè)備的溫度會通過網(wǎng)絡(luò)傳輸?shù)搅饔嬎阋?，我們根?jù)溫度的情況決定是否向管理員發(fā)送告警信息。
假設(shè)某設(shè)備上的溫度傳感器檢測到溫度為105°，此時的設(shè)備系統(tǒng)時間為t0，該讀數(shù)通過網(wǎng)絡(luò)傳輸?shù)诌_流計算引擎后，在t1時刻開始處理該信息。那么，該事件的發(fā)生時間就是，而處理時間就是。 雖然我們總希望發(fā)生時間與處理時間相同，但現(xiàn)實中會由于網(wǎng)絡(luò)通訊、資源調(diào)度、處理模式等等原因造成延遲，所以處理時間總是在發(fā)生時間之后，即t1-t0>0。
注意這里的討論基于一個假設(shè)，即數(shù)據(jù)源時鐘和流計算系統(tǒng)時鐘都是準確的。但在一些實際場景下，發(fā)生時間和處理時間都可能跟實際時間（Wall time）不同，甚至可能出現(xiàn)t1-t0<0的情況。造成這種情況的原因可能是產(chǎn)生數(shù)據(jù)的數(shù)據(jù)源或流計算引擎沒有可靠的時間同步服務(wù)。時間不同步會造成計算結(jié)果錯誤，所以系統(tǒng)設(shè)計者需要慎重考慮，不過這些額外工作不在流計算引擎的考慮范圍。
2.2
流計算應(yīng)用的狀態(tài)和持久化
一個流計算應(yīng)用通常包括多個算子，在定義流計算應(yīng)用狀態(tài)之前需要先理解什么是算子的狀態(tài)。下面給出了一些常見的算子狀態(tài)的例子：
算子在數(shù)據(jù)流中查找某種時序模式（Pattern），需要保存保存一段時間原始數(shù)據(jù)用于匹配；
算子按分鐘/小時/天聚合數(shù)據(jù)，需要保存一些中間結(jié)果；
算子迭代訓練機器學習模型，需要保存當前的模型參數(shù)；
算子在計算時需要參考歷史數(shù)據(jù)，需要保存一些歷史數(shù)據(jù)；
可見，算子的狀態(tài)通常是指為了完成當前的計算而保存的一些基于歷史數(shù)據(jù)產(chǎn)生的內(nèi)容。我們以第一個例子展開說明。假設(shè)我們想在數(shù)據(jù)流中查找下面形狀的模式：
這個模式是由多個數(shù)據(jù)點構(gòu)成，所以算子不能僅根據(jù)當前處理的一條記錄完成模式匹配，而是必須緩存一段歷史數(shù)據(jù)。下圖展示了不同時刻節(jié)點的狀態(tài)，以及隨著狀態(tài)的變化得到的不同匹配結(jié)果。在t0時刻算子狀態(tài)中保存的數(shù)據(jù)與模式不匹配，在t1時間如果后續(xù)抵達的數(shù)據(jù)使算子狀態(tài)如上面的分支，那么此時達成匹配；而如果后續(xù)抵達的數(shù)據(jù)使算子狀態(tài)如下面的分支，則仍然為不匹配。
需要注意的是，算子并非一定有狀態(tài)（state-full），也可以無狀態(tài)（state-less)。例如，我們希望根據(jù)數(shù)據(jù)里的某種屬性進行過濾，而這種屬性跟被處理的數(shù)據(jù)無關(guān)（即不會隨時間發(fā)生變化），那么節(jié)點就不需要記錄任何狀態(tài)信息。所以，算子是否有狀態(tài)僅取決于算子本身的處理邏輯。
下面我們定義流計算應(yīng)用的狀態(tài)。一般情況下，流計算應(yīng)用的狀態(tài)定義為在某個時刻其中每個算子的狀態(tài)的集合。但是，這里隱含一個前提條件——各個算子的狀態(tài)是按數(shù)據(jù)對齊的。下面是按數(shù)據(jù)對齊的一個示例：
假設(shè)流計算應(yīng)用中存在兩個算子o1和o2算子（暫時忽略數(shù)據(jù)源和輸出），算子o2位于算子o1的下游，假設(shè)兩個算子都是有狀態(tài)的。如果o1在處理完數(shù)據(jù)r1和r2之后，狀態(tài)記做s1_1；而算子o1處理完r1和r2之后輸出的結(jié)果記做r1+2，算子o2在處理完之后，狀態(tài)記做s2_1。那么應(yīng)用的狀態(tài)sapp={s1_1,s2_1}，即按照數(shù)據(jù)對齊的每個算子狀態(tài)的集合。相應(yīng)的，在算子o1和o2都處理完數(shù)據(jù)r3和r4之后，應(yīng)用新的狀態(tài)sapp={s1_2,s2_2}。 強制應(yīng)用內(nèi)算子之間按數(shù)據(jù)對齊是為了在狀態(tài)恢復(fù)時可以從統(tǒng)一的起點開始。數(shù)據(jù)的對齊方式可以是數(shù)據(jù)分批，或者在數(shù)據(jù)中插入一些標記（稱為Marker或Barrier），這些具體的策略在閱讀完第四節(jié)對流計算框架的介紹后可以更好理解。
要支持應(yīng)用在故障后不必完全重新計算，還需要對狀態(tài)進行持久化。但持久化有一定的IO、存儲開銷，所以一般情況下不會在處理完每條記錄后都進行持久化，而是按一定的周期，比如每5秒、或每處理1000條記錄等等。在有些計算引擎中，為了減少狀態(tài)持久化的開銷，支持增量的更新策略，即每次持久化只更新與上次不同的部分。另外，由于狀態(tài)有存儲開銷，所以在編寫流計算算子時需要注意不要使算子狀態(tài)無限增長，例如可以引入Time-To-Live（TTL）或某種自定義的清理機制。
2.3
處理模式（Process pattern）
處理模式要解決的是前面提到的“Where”問題，即在什么時間范圍的數(shù)據(jù)上執(zhí)行計算。如果在處理之前我們已經(jīng)得到了所有的數(shù)據(jù)，那么可以一次性處理所有數(shù)據(jù)；而如果數(shù)據(jù)源源不斷到來，情況會復(fù)雜一些，我們需要分一些具體情況來討論。
時間無關(guān)（Time-agnostic）
有時候處理數(shù)據(jù)流并不需要考慮數(shù)據(jù)的發(fā)生時間或處理時間。最典型的一個場景就是“過濾”（Filter），比如我們希望在后續(xù)處理中只考慮滿足某些條件的數(shù)據(jù)（例如溫度大于50°），那么在篩選過程中，我們并不需要考慮數(shù)據(jù)產(chǎn)生或處理時間。注意，這里所謂的“時間無關(guān)”并不一定是指數(shù)據(jù)的時間維度不重要，而是指在處理數(shù)據(jù)時不需要考慮時間。另一個時間無關(guān)的例子，在一些近似學習算法中（如streaming K-means)，每次接收到新的數(shù)據(jù)后迭代得到新的模型。
窗口化（Windowing）
有些情況下在處理數(shù)據(jù)前，需要考慮數(shù)據(jù)的產(chǎn)生時間或處理時間。最簡單的一種場景是按照處理時間定期執(zhí)行計算，例如每隔5分鐘，系統(tǒng)將過去5分鐘內(nèi)抵達的數(shù)據(jù)處理一遍。注意這時候一條數(shù)據(jù)抵達系統(tǒng)后可能不會立刻被處理，而是需要等待一段時間。時間有關(guān)的處理往往都可以描述成一種“窗口化”的形式，而每一個窗口代表從時間維度來看，哪一段數(shù)據(jù)是計算的對象，比如例子里提到的每隔5分鐘處理過去5分鐘內(nèi)抵達的數(shù)據(jù)。
窗口一般有兩個要素：
起始時間，指窗口內(nèi)數(shù)據(jù)的開始時間，比如8:00、8:05、8:10...；
長度，指窗口在時間上的跨度，比如5分鐘；
前面提到了兩種時間概念：發(fā)生時間和處理時間。于是，窗口化也大都按照這兩種時間來考慮?；谔幚頃r間的窗口化比較容易理解，比如每隔n分鐘，處理過去m分鐘積累的數(shù)據(jù)。如果n=m，那么時間窗口之間是沒有重疊（non-overlapping）的；而如果m>n，則連續(xù)兩次處理的數(shù)據(jù)是有一定重疊（overlapping）的。
基于發(fā)生時間的窗口化要稍微復(fù)雜一些（如上圖下部所示）。以設(shè)備高溫報警為例，我們希望按照發(fā)生時間來統(tǒng)計機器每5分鐘內(nèi)的平均溫度。那么可能的時間窗口為
..., 8:00 ~ 8:05, 8:05 ~ 8:10, 8:10~8:15, ...
數(shù)據(jù)是有延遲地、亂序地抵達的，所以在8:05時刻是無法得到8:00~8:05所有數(shù)據(jù)的。因為延遲總是存在的，所以一般計算要延后一些。那么需要延后多少呢？實際上，這個問題如果脫離真實場景是無法回答的。如果傳感器讀數(shù)可以從傳感器通過網(wǎng)絡(luò)直接發(fā)送到處理系統(tǒng)，那么這里的延遲可能只有數(shù)毫秒（在網(wǎng)絡(luò)無故障的前提下）；但如果傳感器無法聯(lián)網(wǎng)，先緩存在機器上，需要人工每天用硬盤拷貝后傳輸?shù)教幚硐到y(tǒng)，那么延遲可能是數(shù)十小時。
除了上述兩種典型窗口化方式以外，還可以有一些自定義的窗口定義方式。一個最典型的例子是分析人們登錄一個網(wǎng)站的行為——從登錄到登出之間存在一個session，而若需要按照session來進行計算就需要將某個用戶ID的每個session定義為一個窗口。窗口化的形式看似多樣，其本質(zhì)是在定義被計算數(shù)據(jù)所處的時間范圍，即解決Where的問題。這里在討論窗口化時，假定待處理數(shù)據(jù)在時間上是連續(xù)的，即窗口一般是一段連續(xù)的時間——這并非一種硬性要求，流計算引擎通常會提供一些自定義的實現(xiàn)方式。
窗口化處理模式意味著算子一定是有狀態(tài)的（即state-full），并且狀態(tài)有有限的生命周期。首先，窗口化意味著必須緩存一定數(shù)據(jù)或計算中間結(jié)果，以便在窗口內(nèi)數(shù)據(jù)完整抵達后執(zhí)行計算，所以一定是有狀態(tài)的。然后，在窗口化的情況下，狀態(tài)總是與某個窗口綁定的，一旦窗口內(nèi)數(shù)據(jù)完整抵達并完成計算，狀態(tài)就不再有意義，所以隨著窗口生命期的完結(jié)，狀態(tài)的生命期也隨之結(jié)束。但是需要注意的是，并非所有算子狀態(tài)都是有生命期的，例如我們想統(tǒng)計所有接收到的設(shè)備溫度數(shù)據(jù)里的最大值，這時候狀態(tài)就不會有明確的生命期限。最后，有狀態(tài)的算子并不一定存在窗口化，想想前面提到的迭代訓練機器學習模型的例子。
窗口生命期管理
我們來考慮一個窗口的生命周期。理解了這個問題可以方便我們了解流計算引擎的算子狀態(tài)管理。流計算引擎的內(nèi)部狀態(tài)一般存儲在參與計算節(jié)點的內(nèi)存或磁盤，意味著一定的存儲開銷?；谔幚頃r間的窗口生命期管理很簡單，因為這與被處理的數(shù)據(jù)流無關(guān)，流計算引擎只需要根據(jù)系統(tǒng)時間定期新建或銷毀窗口。
基于發(fā)生時間的窗口生命期管理要復(fù)雜一些。沿用前面計算窗口內(nèi)平均溫度的例子，考慮8:00~8:05這個窗口。由于數(shù)據(jù)是陸續(xù)抵達的，為了計算平均溫度就必須緩存兩個值，即落在該窗口內(nèi)的所有溫度讀數(shù)的個數(shù)和它們之和，最終可以計算得到平均數(shù)：
窗口的生命期內(nèi)引擎需要保存這兩個值（即算子的狀態(tài)），直到計算得到溫度平均數(shù)為止。那么這個窗口是什么時候創(chuàng)建的呢？是在引擎接收到第一個落在該時間區(qū)間內(nèi)的值的時候。根據(jù)前面的討論我們知道，這個時間的具體值是與應(yīng)用場景相關(guān)的。如下圖所示，假設(shè)在9:00時刻接收到落在8:00~8:05窗口內(nèi)的第一條記錄，則此刻創(chuàng)建窗口。隨著后續(xù)數(shù)據(jù)抵達不斷更新窗口狀態(tài)。
那么窗口會在什么時刻關(guān)閉呢？關(guān)閉窗口意味著完成計算并清空狀態(tài)，嚴格意義上講需要等窗口內(nèi)數(shù)據(jù)都抵達后才能關(guān)閉窗口，但這個問題同樣與應(yīng)用場景相關(guān)。前面假定我們接收到一個窗口內(nèi)讀數(shù)的時刻是在9:00，說明延遲約1小時，那么在9:05時刻我們可能認為所有在8:00~8:05分的數(shù)據(jù)都已經(jīng)被接收了，于是可以計算平均溫度并關(guān)閉該窗口；或者，我們出于保守的目的，認為需要等到9:10才能確認數(shù)據(jù)完全接收。注意，無論我們最終選擇9:05還是9:10，這都只是我們根據(jù)經(jīng)驗人為設(shè)定的一個估計值，并不能確保在此之前所有窗口內(nèi)的數(shù)據(jù)都已經(jīng)到達。當然，如果某個實時系統(tǒng)能夠給出延遲的上界，我們就可以按照上界給出一個絕對可靠的時間。所以，流計算引擎并不能保證計算結(jié)果與事實一致，因為確保所有數(shù)據(jù)在給定時間之前抵達并非計算引擎需要考慮的問題，計算引擎要做的是提供完備的描述語義。
為了描述這種關(guān)閉窗口的時間，流計算引擎引入了一個稱為水位線（Watermark）的概念。水位線實際上是一個映射關(guān)系，即根據(jù)當前系統(tǒng)的狀態(tài)估計數(shù)據(jù)抵達情況。在前面的例子里，如果我們根據(jù)當前時間是9:10判斷在8:05之前所有數(shù)據(jù)都已經(jīng)抵達，則映射的輸入是“當前時間9:10”，輸出是“8:05”。當然也可以用其它狀態(tài)來推測水位線，比如如果我們接收到8:05時刻的溫度讀數(shù)，則認為8:05之前所有數(shù)據(jù)已經(jīng)抵達。并不存在完美普適的估計水位線的方法，而需要根據(jù)應(yīng)用場景設(shè)定最適合的估計方法。
水位線一旦超過某個時間窗口的最大時間，則可以計算出窗口結(jié)果并關(guān)閉窗口。窗口關(guān)閉后，引擎就可以釋放該窗口所占用的存儲空間。在前面的例子里，計算平均值只需要保存兩個值，但在有些場景下窗口內(nèi)可能需要保存原始數(shù)據(jù)，那樣的話對存儲的占用開銷會比較大。所以，準確地估計水位線非常重要。通常情況下，窗口保存的時間越短則存儲代價越小、計算結(jié)果越粗糙；反之，則存儲代價越大、計算結(jié)果越精確。計算引擎需要開發(fā)者來平衡開銷和精度。
在一些流計算引擎里，除了水位線外還會有最大延遲的概念，即在水位線已經(jīng)超過窗口最大時間后還需要維持一段時間。在某些場景下可能會使窗口定義更加靈活，但從窗口的生命期角度來說，最大延遲也只是水位線的一部分。
2.4
窗口的計算時機
對某個時間窗口而言，在水位線超過窗口最大時間以后就可以計算得到該窗口對應(yīng)的值。但在此之前，有時候我們也希望提前處理已經(jīng)落在窗口內(nèi)的數(shù)據(jù)，以便提前獲得一些估計結(jié)果，這時候就需要用到觸發(fā)器（Trigger）的概念。觸發(fā)器決定了窗口計算的時機，例如，可以在窗口初始化之后，每隔一段時間計算一次中間值，或每接收到一定數(shù)量的數(shù)據(jù)后計算一次中間值；或者在接收到特定數(shù)據(jù)時計算一次中間值。當然，一般的計算引擎都對每個窗口自動注冊一個默認觸發(fā)器——在窗口關(guān)閉時計算一次。下圖展示了隨著水位線超過窗口最大時間后計算溫度平均值的例子。
最終每個窗口只會保留一個計算結(jié)果，所以每次計算都涉及到如何更新窗口結(jié)果的問題。一般有兩種更新方式：替換和累積。其中替換是指后一次的計算結(jié)果替換前一次的結(jié)果；累積是指后一次的計算結(jié)果與前一次的計算結(jié)果經(jīng)過某種運算（如加法）得到新的結(jié)果。在前面關(guān)于計算5分鐘內(nèi)平均溫度的例子里，如果我們定義的觸發(fā)方式是每分鐘（處理時間）計算一次中間結(jié)果，那么對于窗口8:00~8:05，我們依次得到的結(jié)果是：
處理時間 | 9:01 | 9:02 | 9:03 | 9:04 | 9:05 | 9:06 |
中間結(jié)果 | 91° | 95° | 97° | 97° | 97° | 95° |
窗口的第一次計算是在9:01，每次計算中間結(jié)果，都可以根據(jù)目前窗口內(nèi)的所有數(shù)據(jù)來計算，所以可以替換掉前面已有的中間結(jié)果。而窗口8:00~8:05最終得到的計算結(jié)果在9:06。通過上述計算過程，我們不僅可以得到最終結(jié)果，還能觀察到中間結(jié)果的變化過程。累積的更新形式往往用于那些可以分段計算的情況，比如我們可以需要統(tǒng)計一段時間內(nèi)接收到數(shù)據(jù)的量，就可以將后續(xù)統(tǒng)計結(jié)果疊加在之前的中間結(jié)果之上。
2.5
流計算引擎的一致性
在前面關(guān)于Lambda結(jié)構(gòu)的討論中，曾經(jīng)提到實時計算給人的印象是“不準確”，這種不準確的根源是一致性問題。一般來說，針對流數(shù)據(jù)的計算一致性可以分為三種類型：
At-most-once：可以理解為“盡力而為”。一條記錄可能在流計算系統(tǒng)中丟失（例如網(wǎng)絡(luò)抖動），也可能被重復(fù)處理。
At-least-once：比起At-most-once，這種語義保證每條記錄至少會被處理一次，即不會發(fā)生數(shù)據(jù)丟失，但一條記錄仍然可能被重復(fù)處理多次。
Exactly-once：這種語義保證記錄不會丟失，并且每條記錄對狀態(tài)的影響只能有一次。
要理解上面的三種語義，首先要清楚這里提到的“丟失”和“重復(fù)”特指在流計算系統(tǒng)內(nèi)部發(fā)生的行為，而不是指數(shù)據(jù)本身存在的問題。比如，數(shù)據(jù)在到達流計算引擎之前可能已經(jīng)丟了，或數(shù)據(jù)本身就包含一定的重復(fù)率，那這些問題并不在流計算引擎考慮之內(nèi)。那么，為什么流計算系統(tǒng)內(nèi)會發(fā)生數(shù)據(jù)丟失或重復(fù)呢？這是因為流計算系統(tǒng)通常包含多個節(jié)點（如下圖示例），節(jié)點間通過網(wǎng)絡(luò)通信，并且節(jié)點本身也可能發(fā)生故障，導(dǎo)致數(shù)據(jù)重復(fù)或丟失。
從正確性的角度來看，At-most-once是不可用的，因為它無法保證計算結(jié)果是否能反映數(shù)據(jù)的情況。假設(shè)我們希望統(tǒng)計一個網(wǎng)站每小時的瀏覽次數(shù)，在At-most-once下，每個小時得到的統(tǒng)計結(jié)果可能比真實的情況多（因為數(shù)據(jù)被重復(fù)處理），也可能少（因為數(shù)據(jù)被丟失）。
At-least-once確保不會發(fā)生數(shù)據(jù)丟失。假設(shè)我們利用流計算來監(jiān)控某項指標，一旦監(jiān)測到某種情況則發(fā)出告警信號。如果一些數(shù)據(jù)被重復(fù)處理，可能會發(fā)出重復(fù)的告警，但因為數(shù)據(jù)不會被丟失，所以確保在出現(xiàn)問題時，管理員能收到告警。因此，在一些對數(shù)據(jù)精確性要求不高的場景下At-least-once仍然可以較好的勝任。
Exactly-once是一個容易讓人產(chǎn)生誤解的概念。這種誤解通常來自兩方面：
誤解1：支持Exactly-once語義的流計算引擎對每條記錄只處理一次；
誤解2：支持Exactly-once語義只是流計算引擎內(nèi)部的行為，與外部系統(tǒng)無關(guān)
只有解開上面的兩個誤解，才能真正理解什么才是Exactly-once。下面我們分別討論。我們用一個類似前面計算一定時間窗口內(nèi)平均溫度的例子，這次我們計算的是一個時間窗口內(nèi)所有記錄的和，比如我們想知道每小時瀏覽一個網(wǎng)站的人的數(shù)量。我們得到的原始記錄可能如下：
[2019-03-28 11:01:35] 4
[2019-03-28 11:17:13] 17
[2019-03-28 11:23:35] 2
...
括號里是日志時間，后面的整數(shù)代表從上一次統(tǒng)計到這一次之間的人數(shù)?，F(xiàn)在我們想知道11點以內(nèi)究竟有多少人數(shù)瀏覽了網(wǎng)頁。為了方便討論，我們先忽略底層引擎的實現(xiàn)細節(jié)。而為了計算一個小時內(nèi)所有數(shù)據(jù)之和，就必須在狀態(tài)中保留已經(jīng)接收到的所有數(shù)據(jù)之和，直到11點對應(yīng)的窗口關(guān)閉為止。但是，如果在窗口關(guān)閉之前（比如在11:30的時候）計算節(jié)點斷電了呢？
這會引發(fā)一個問題——在11:30之前已經(jīng)計算得到的中間結(jié)果是不是丟失了？一般情況下，節(jié)點都會把中間結(jié)果定期保存到持久化存儲里（一般是支持多備份的分布式文件系統(tǒng)），但不會處理完每條記錄都這么做，因為那樣的開銷太大。假設(shè)節(jié)點每10分鐘持久化一次，在節(jié)點斷電之前最后一次持久化的時間是11:20。那在節(jié)點再次啟動（或者計算遷移到其它節(jié)點）之后，我們能讀到的狀態(tài)仍然在11:20。那么從11:20~11:30這段時間內(nèi)的數(shù)據(jù)怎么辦？這些數(shù)據(jù)在斷電之前已經(jīng)被引擎處理過一次了。
這個問題并不是流計算引擎能獨立回答的，因為它無法緩存這些原始數(shù)據(jù)。所以，Exactly-once語義要求數(shù)據(jù)源有可重放（Replayable）的特性。在計算重新開始時，流計算引擎要求數(shù)據(jù)源再次從11:20開始消費數(shù)據(jù)，這樣就能與持久化的狀態(tài)銜接上。但這樣的話，有些數(shù)據(jù)會被消費兩次（或更多），比如一條記錄在11:25，可能在前一次節(jié)點斷電前被處理過一次，而恢復(fù)后又處理了一次。但是，這條記錄對最終統(tǒng)計的總和來講只記錄了一次。
回到前面提到的兩種誤解。Exactly-once的語義并不是保證每條數(shù)據(jù)只被處理了一次，而是確保一條數(shù)據(jù)對最終結(jié)果只會影響一次。此外，Exactly-once也不僅是流計算引擎的行為，還需要外部系統(tǒng)配合完成。前面的例子里只提到了數(shù)據(jù)源的要求，實際上對輸出系統(tǒng)可能也有要求。試想流計算系統(tǒng)接收源源不斷的數(shù)據(jù)，處理后持續(xù)不斷寫出到外部存儲（例如一個數(shù)據(jù)庫），如果在某個時間點輸出節(jié)點斷電了，則恢復(fù)后斷電前臨時寫出的結(jié)果需要被回滾。通過閱讀Flink里Exactly-once語義的實現(xiàn)方式能更好地理解上面所說的問題；此外，Kafka也有相關(guān)的討論，由于Kafka常常是流計算引擎的入口和出口，其本身支持尤其重要。
一致性是流計算引擎最重要的屬性之一。在根據(jù)應(yīng)用場景選擇流計算引擎之前要確認究竟需要什么樣的一致性要求：At-most-once、At-least-once還是Exactly-once。At-most-once常常意味著結(jié)果完全不可控；At-least-once對一些可以容忍重復(fù)的應(yīng)用場景是不錯的選擇；Exactly-once從正確性上最完備，但也意味著更多的開銷。有些流計算引擎實際上提供關(guān)于一致性的配置，可以自由選擇，由用戶自己來平衡性能和代價。
下期預(yù)告
3
流計算框架比較
3.1
Structured Streaming
3.2
Flink
3.3
Kafka Streams
3.4
流計算框架介紹小結(jié)
數(shù)莓派
以數(shù)為媒，共建工業(yè)大數(shù)據(jù)社區(qū)
