劉陽，張揚揚，周號益

面向流式數(shù)據(jù)處理系統(tǒng)的高效故障恢復(fù)方法

劉陽1，2，3，張揚揚1，2，周號益1，2，4*

（1.北京航空航天大學(xué) 大數(shù)據(jù)科學(xué)與腦機智能高精尖創(chuàng)新中心，北京 100191； 2.北京航空航天大學(xué) 計算機學(xué)院，北京 100191； 3.北京航空航天大學(xué) 未來空天技術(shù)學(xué)院/高等理工學(xué)院，北京 100191； 4.北京航空航天大學(xué) 軟件學(xué)院，北京 100191）（?通信作者電子郵箱haoyi@buaa.edu.cn）

針對流式數(shù)據(jù)處理系統(tǒng)Flink無法高效處理單點故障的問題，提出了一種基于增量狀態(tài)和備份的故障容錯系統(tǒng)Flink+。首先，提前建立備份算子和數(shù)據(jù)通路；然后，對數(shù)據(jù)流圖中的輸出數(shù)據(jù)進行緩存，必要時使用磁盤；其次，在系統(tǒng)快照時進行任務(wù)狀態(tài)同步；最后，在系統(tǒng)故障時使用備份任務(wù)和緩存的數(shù)據(jù)恢復(fù)計算。在系統(tǒng)實驗測試中，F(xiàn)link+在無故障運行時沒有顯著增加額外容錯開銷；而在單機和分布式環(huán)境下處理單點故障時，與Flink系統(tǒng)相比，所提系統(tǒng)在單機8任務(wù)并行度下故障恢復(fù)時間減少了96.98%，在分布式16任務(wù)并行度下故障恢復(fù)時間減少了88.75%。實驗結(jié)果表明，增量狀態(tài)和備份方法一起使用可以有效減少流式系統(tǒng)單點故障的恢復(fù)時間，增強系統(tǒng)的魯棒性。

流式數(shù)據(jù)處理系統(tǒng)；故障恢復(fù)；分布式檢查點；狀態(tài)備份；Apache Flink

0 引言

大數(shù)據(jù)時代，隨著互聯(lián)網(wǎng)、物聯(lián)網(wǎng)等技術(shù)的快速發(fā)展，諸如工業(yè)監(jiān)控、社交媒體、實時搜索引擎等應(yīng)用場景產(chǎn)生了海量的數(shù)據(jù)，并對計算處理有了更嚴(yán)格的要求，需要數(shù)據(jù)處理系統(tǒng)提供低延遲的實時計算，而對實時計算的需求進一步促進了分布式數(shù)據(jù)處理系統(tǒng)從批處理（Batch Processing）模式逐步轉(zhuǎn)變?yōu)榱魈幚恚⊿tream Processing）模式。

批處理系統(tǒng)通過對輸入數(shù)據(jù)進行采樣收集，當(dāng)數(shù)據(jù)規(guī)模達(dá)到設(shè)定閾值后利用批處理引擎計算累積的數(shù)據(jù)，可以反映一段時間內(nèi)數(shù)據(jù)的特征，同時還能夠保證數(shù)據(jù)分析結(jié)果的正確性，也被廣泛用于機器學(xué)習(xí)、圖計算等領(lǐng)域，并采用分布式檢查點［1］、反應(yīng)式故障恢復(fù)［2］等技術(shù)進行容錯。此外，批處理系統(tǒng)的框架相對簡單且易于擴展。雖然批處理可以達(dá)到很高的吞吐，但在實時性方面難以滿足當(dāng)前大數(shù)據(jù)背景下各類實時應(yīng)用的低延遲的需求。

相較于批處理模式，流式系統(tǒng)逐條地對輸入數(shù)據(jù)進行實時處理，可以捕獲動態(tài)實時數(shù)據(jù)的最新特征，更快地挖掘數(shù)據(jù)背后的價值。流式系統(tǒng)通常會對無界數(shù)據(jù)提供長期穩(wěn)定且實時的計算處理，可以很好地滿足商業(yè)公司對數(shù)據(jù)處理的實時性需求。流式處理分為有狀態(tài)處理和無狀態(tài)處理，前者語義豐富可以表達(dá)更真實的數(shù)據(jù)場景，后者的使用場景較為簡單。在分布式系統(tǒng)中，有狀態(tài)的流數(shù)據(jù)處理完全依賴于前序計算狀態(tài)，如果在數(shù)據(jù)處理過程中發(fā)生故障，將導(dǎo)致前序狀態(tài)的丟失，從而導(dǎo)致流計算必須整體重新開始，嚴(yán)重情況下如持續(xù)大規(guī)模數(shù)據(jù)輸入場景，甚至無法實現(xiàn)狀態(tài)恢復(fù)，故障后系統(tǒng)恢復(fù)代價很高。因此，針對有狀態(tài)流數(shù)據(jù)處理的故障容錯需要系統(tǒng)在計算資源和處理速度上進行一定妥協(xié)。但隨著新興應(yīng)用場景的出現(xiàn)，流式系統(tǒng)對數(shù)據(jù)處理延遲和備份開銷要求越來越嚴(yán)格，當(dāng)前故障容錯機制面臨著新的挑戰(zhàn)。

目前，主流開源流式系統(tǒng)［3-11］和商業(yè)流式系統(tǒng)［12］根據(jù)系統(tǒng)設(shè)計與能力支持的不同，采用的容錯機制不盡相同，針對的應(yīng)用場景和容錯能力也有較大差異。如Storm［10-11］采用消息確認(rèn)的機制實現(xiàn)容錯，但僅支持至少一次語義支持；Spark Streaming［5-9］將流式數(shù)據(jù)視為一個個小的批數(shù)據(jù)，利用微批（Micro Batch）以支持流式計算，并復(fù)用Spark的血緣容錯機制，支持精確一次語義，但對其他的流式特性支持較差。Flink［3-4］采用了一種簡化的Chandy?Lamport分布式快照算法［13-14］，在保證精確一次語義的同時，實現(xiàn)輕量級的容錯。當(dāng)一次快照完成之后可視為數(shù)據(jù)成功處理，從而實現(xiàn)端到端的精確一次語義支持。當(dāng)故障發(fā)生時，所有計算任務(wù)整體回滾到上一次快照狀態(tài)，并重新消費之前的數(shù)據(jù)。然而這種方法存在一個弊端，即使故障規(guī)模較小，甚至是單節(jié)點故障，也不得不將所有計算節(jié)點進行回滾，故障恢復(fù)時間長，并且需要重新計算因回滾丟失的進度。

針對這一問題，本文提出一種基于增量式狀態(tài)備份的高效故障恢復(fù)系統(tǒng)Flink+，通過增量式狀態(tài)同步實現(xiàn)快速狀態(tài)備份；利用上游輸出緩存和備份數(shù)據(jù)通路實現(xiàn)故障任務(wù)的快速恢復(fù)。Flink+基于Flink已有的分布式快照機制，利用已有的計算任務(wù)進行增量式的互相備份，一個計算任務(wù)在進行自己主計算的同時，也負(fù)責(zé)備份其他節(jié)點的任務(wù)狀態(tài)和計算邏輯。在創(chuàng)建計算圖時，備份任務(wù)提前建立好與主任務(wù)上下游任務(wù)的網(wǎng)絡(luò)連接，以降低故障恢復(fù)時任務(wù)切換的時間，該備份連接在無故障運行時并不進行傳輸；當(dāng)快照時，備份任務(wù)增量式地同步主任務(wù)的狀態(tài)，降低狀態(tài)備份開銷。在故障發(fā)生后，備份任務(wù)立即啟動備份計算邏輯，并利用備份網(wǎng)絡(luò)連接接管上下游數(shù)據(jù)，實現(xiàn)快速的故障任務(wù)切換與恢復(fù)。為了進一步細(xì)化故障恢復(fù)的粒度，采用上游輸出緩存機制，在無故障運行時，上游任務(wù)的輸出會被保留一段時間，以防止下游故障之后需要從數(shù)據(jù)源頭進行重計算，只需要重新消費其上游的輸出數(shù)據(jù)，進一步降低故障恢復(fù)時間。

本文的主要工作如下：

1）設(shè)計了一種基于增量式狀態(tài)的快速備份方法，結(jié)合快照機制和增量狀態(tài)備份，實現(xiàn)對系統(tǒng)狀態(tài)的快速備份。

2）利用上游輸出緩存和備份數(shù)據(jù)通路實現(xiàn)故障任務(wù)的快速切換和狀態(tài)恢復(fù)，提高系統(tǒng)對單點故障的處理速度。

3）在開源流式系統(tǒng)Flink中進行了實現(xiàn)和實驗驗證，實驗結(jié)果驗證了本文方法的可行性和有效性，該方法在無故障運行時沒有顯著增加額外容錯開銷，同時實現(xiàn)了非常顯著的故障恢復(fù)加速效果，加速比可達(dá)6～8。

1 研究背景

1.1　語義支持

在流式系統(tǒng)中，數(shù)據(jù)的最小單位是消息，對消息的處理次數(shù)保證被稱為投遞語義（Delivery Semantic），為了更好地理解流式計算，需要首先介紹一下流處理的三種語義：

1）最多一次：對于數(shù)據(jù)中的每條消息，至多只進行一次處理，如果發(fā)生故障，消息會丟失，并且系統(tǒng)不會進行故障恢復(fù)計算，而是繼續(xù)處理后續(xù)到達(dá)的消息。這種語義放松了系統(tǒng)的計算保證，簡化了系統(tǒng)設(shè)計，適合能夠容忍數(shù)據(jù)丟失的應(yīng)用。

2）至少一次：對于每條消息，至少會計算一次。考慮消息已被計算但未被確認(rèn)的情況，若此時發(fā)生故障，系統(tǒng)重啟，數(shù)據(jù)源重新發(fā)送未被確認(rèn)的消息，則會導(dǎo)致對同一條消息的多次計算。這種語義能夠保證數(shù)據(jù)的完整性，但需要上層應(yīng)用處理數(shù)據(jù)重復(fù)的問題。

3）精確一次：不論是發(fā)生故障還是正常運行，對于每個消息，從系統(tǒng)整體的端到端來看都只會處理一次，即一份輸入數(shù)據(jù)對應(yīng)一份輸出數(shù)據(jù)。這種語義提供最強的數(shù)據(jù)保證，可以滿足對數(shù)據(jù)有強一致性要求的應(yīng)用，但增加了流式系統(tǒng)的復(fù)雜性。

1.2　容錯相關(guān)工作

流式系統(tǒng)的容錯機制［15-16］可分為三個類別：主動備份、被動備份以及混合備份［17-18］。

1）主動備份：系統(tǒng)中的計算節(jié)點都有一個獨立的備份節(jié)點，備份節(jié)點和主節(jié)點擁有一樣的資源和計算邏輯，系統(tǒng)正常運行時，二者處理一樣的數(shù)據(jù)流，因此主節(jié)點和備份節(jié)點的狀態(tài)可以實現(xiàn)同步，當(dāng)然備份節(jié)點的輸出會被丟棄或者緩存起來，這取決于不同的實現(xiàn)方式，只有主節(jié)點的輸出會傳向下游計算節(jié)點。當(dāng)系統(tǒng)中發(fā)生節(jié)點故障時，自動切換到備份節(jié)點，因為主節(jié)點和備份節(jié)點狀態(tài)是同步的，只有切換時間代價，沒有狀態(tài)恢復(fù)的時間代價。該策略可以實現(xiàn)最低延遲的故障處理，但僅考慮單一節(jié)點備份就需要消耗2倍的軟硬件資源，對于大型系統(tǒng)來說負(fù)擔(dān)較重。雙集群備份是一種典型的主動備份方法。

2）被動備份：對于系統(tǒng)中的每一個節(jié)點，定期將節(jié)點的狀態(tài)通過快照或者其他形式保存到備份節(jié)點上，備份節(jié)點可以獲取主節(jié)點的計算邏輯，當(dāng)系統(tǒng)發(fā)生故障時，備份節(jié)點從最近一次緩存的狀態(tài)開始恢復(fù)，輸入數(shù)據(jù)可以一起放入快照或者緩存在輸出隊列中。該策略的具體的實現(xiàn)方式較多，但是不可避免的是節(jié)點狀態(tài)存在恢復(fù)過程，可以滿足對故障恢復(fù)時間要求不嚴(yán)格的場景需求，但仍無法滿足諸如“雙十一”等關(guān)鍵業(yè)務(wù)的嚴(yán)格需求。被動備份策略往往采用節(jié)點冷啟動的方案，因此資源需求相對較低。分布式檢查點是一種典型的被動備份方法。

3）混合備份［19-22］：該策略綜合考慮了上述不同策略的優(yōu)缺點，通過部分熱啟動的備份節(jié)點實時進行狀態(tài)同步。系統(tǒng)正常工作時，備份節(jié)點會對同步的狀態(tài)進行恢復(fù)計算，發(fā)生故障時，主節(jié)點計算邏輯就可以直接切換到備份邏輯上。例如，相關(guān)工作FP4S（Fragment?based Parallel State Recovery for Stateful Stream Applications）［23］借鑒鏈?zhǔn)綇?fù)制的思想將任務(wù)組織成環(huán)形一致性哈希（Consistent Hashing）進行互備，建立路由和鄰居表來選擇狀態(tài)的地理優(yōu)先備份，減少網(wǎng)絡(luò)延遲；同時使用糾錯碼將任務(wù)內(nèi)存狀態(tài)分塊寫入備份節(jié)點，在恢復(fù)時并行拉取狀態(tài)塊來提高恢復(fù)速度，其后續(xù)工作SR3（Customizable Recovery for Stateful Stream Processing Systems）［24］更進一步優(yōu)化了狀態(tài)備份的選擇性。

混合策略可以有效結(jié)合各種容錯機制的優(yōu)點，充分利用資源降低開銷，加速故障恢復(fù)。而開源流式系統(tǒng)目前還是使用單一容錯策略居多，因此存在進一步優(yōu)化的空間。Apache Storm作為最早的流式系統(tǒng)，是原生的流式系統(tǒng)，采用被動備份和消息確認(rèn)機制來實現(xiàn)容錯，但只能提供至少一次投遞語義，容錯能力薄弱，且消息確認(rèn)機制導(dǎo)致系統(tǒng)吞吐量不高；Apache Spark的Spark streaming部分屬于流處理框架，底層采用彈性分布式數(shù)據(jù)集（Resilient Distributed Dataset， RDD）來實現(xiàn)流計算，但本質(zhì)上是“微批”的處理思想，在處理延遲上存在不足（盡管Spark Streaming的底層已經(jīng)開始向流處理原生框架遷移，但是在適配方面還是略微遜色）；Apache Flink目前是最火熱的開源流處理框架之一，原生流處理框架使Flink可以實現(xiàn)毫秒級的事務(wù)，并且相比Storm又具有更高的吞吐量，但是Flink本身的容錯機制也存在一些問題，因此，本文針對Flink系統(tǒng)，提出了一種混合備份方法，設(shè)計并實現(xiàn)了Flink+系統(tǒng)，通過主動狀態(tài)同步與被動計算來實現(xiàn)高效的故障恢復(fù)。

2 Flink

Apache Flink屬于原生的流處理架構(gòu)，但Flink也同時支持批處理，它把批處理當(dāng)作流處理中的一種特殊情況，用流處理來模擬批處理，本文在此只討論Flink的流處理框架。

在Flink中，所有的數(shù)據(jù)都被看作流的一部分，這種抽象很接近于現(xiàn)實世界?；ヂ?lián)網(wǎng)數(shù)據(jù)往往都是事件流，從一個數(shù)據(jù)庫轉(zhuǎn)移到另外一個數(shù)據(jù)庫，進行一些操作，生成新的事件等，每個事件還往往伴隨著其被處理的時間，而傳統(tǒng)的批處理對時間信息不敏感，因此無法通過時間信息獲取更多數(shù)據(jù)信息，這在電商經(jīng)濟中尤為凸顯，用戶點擊和購買對于商家的推薦有明顯的影響，并且推薦要實時，否則就可能錯失用戶。Flink作為最新一代的原生流處理框架，在事件處理時延上可以實現(xiàn)毫秒級別的延遲，其穩(wěn)定性和可擴展性非常適合大規(guī)模集群的數(shù)據(jù)處理，本文的實驗也基于Flink進行。

2.1　Flink系統(tǒng)架構(gòu)

Flink流處理框架由兩類運行時進程構(gòu)成：JobManager和TaskManager。在后續(xù)討論過程中將前者稱之為作業(yè)管理器，后者稱之為執(zhí)行管理器。

作業(yè)管理器負(fù)責(zé)協(xié)調(diào)申請各種資源、對流處理任務(wù)建模、執(zhí)行快照等。調(diào)度器（Scheduler）負(fù)責(zé)調(diào)度子任務(wù)（SubTask）的運行，快照協(xié)調(diào)器（Checkpoint Coordinator）負(fù)責(zé)分布式快照相關(guān)的邏輯。執(zhí)行管理器是任務(wù)實際運行的地方，執(zhí)行管理器中的任務(wù)槽（Task Slot）作為最小任務(wù)執(zhí)行單位，由作業(yè)管理器申請使用。每個Task Slot同一時刻只能處理一個任務(wù)。不同的Task Slot通過本地消息隊列或者網(wǎng)絡(luò)傳遞數(shù)據(jù)。Flink采用Akka作為底層高并發(fā)的運行時，各組件通過Actor模型通信。

2.2　Flink分布式快照

接下來詳細(xì)介紹Flink目前的容錯恢復(fù)機制［1］。Flink的容錯機制屬于被動備份策略，主要通過分布式快照實現(xiàn)，周期性地把系統(tǒng)算子狀態(tài)備份到遠(yuǎn)端持久化存儲，當(dāng)系統(tǒng)發(fā)生故障后，利用上一次快照的狀態(tài)重新計算并恢復(fù)狀態(tài)。

Flink的分布式快照算法是基于Chandy?Lamport算法［11］修改后實現(xiàn)的異步算法［12］。Chandy?Lamport算法用來對分布式系統(tǒng)狀態(tài)做快照，把分布式系統(tǒng)的全局狀態(tài)和鏈路狀態(tài)記錄下來用以故障恢復(fù)或死鎖檢測等。Flink對其進行修改后，實現(xiàn)了一個輕量級的異步分布式快照算法。

Flink流計算模型中包含Source operator作為數(shù)據(jù)輸入源算子、Transformation operator作為變換算子和Sink operator作為數(shù)據(jù)輸出算子三種算子類型。

Flink系統(tǒng)由一個快照協(xié)調(diào)器協(xié)調(diào)全局狀態(tài)快照，該協(xié)調(diào)器會周期性觸發(fā)快照消息給Source operator，該算子在收到消息后，會首先把自身狀態(tài)備份到快照中，然后生成一個barrier消息傳遞給下游全部算子。該barrier消息標(biāo)有對應(yīng)的序號并且和普通數(shù)據(jù)共享同一通路，當(dāng)下游算子處理到barrier時，會觸發(fā)自身的快照，如果算子有多個輸入通道，那么當(dāng)每個輸入通道都接收到對應(yīng)的barrier后才會觸發(fā)算子的狀態(tài)快照，否則輸入通道會暫時阻塞該通道的輸入數(shù)據(jù)直到快照開始。下游算子把狀態(tài)備份到快照后同樣生成新的barrier，并且傳遞給自己的全部下游算子；當(dāng)Sink operator收到輸入通道的全部對應(yīng)一致序號的barrier后，本次快照完全結(jié)束，并且通知快照協(xié)調(diào)器。之后由協(xié)調(diào)器把備份的狀態(tài)發(fā)送到持久化存儲中。

Flink的這種異步快照方式可以在不影響其他算子正常計算的情況下完成整個執(zhí)行圖的狀態(tài)備份，將快照和數(shù)據(jù)流處理完美地融合在一起。

針對上述提到的阻塞情況，如果因為系統(tǒng)負(fù)載不均，導(dǎo)致算子某個輸入通道被阻塞較長時間會對算子計算造成影響，因此Flink對這種情況設(shè)計了非對齊的快照算法：即當(dāng)?shù)谝粋€輸入通道收到barrier后，就立即向下游算子廣播該barrier；同時立即開始狀態(tài)備份，還會把還沒有接收到對應(yīng)標(biāo)號的輸入通道的數(shù)據(jù)全部備份起來。這種情況下的快照算法需要算子記錄輸入流的部分信息，與原始Chandy? Lamport算法較為相似。這種非對齊的快照會導(dǎo)致備份狀態(tài)的數(shù)據(jù)量變大，同時在狀態(tài)恢復(fù)時需要對狀態(tài)和保存的輸入流數(shù)據(jù)重新計算，增加了恢復(fù)時間。

當(dāng)發(fā)生單一節(jié)點故障時，F(xiàn)link系統(tǒng)會計算故障區(qū)域，因為流計算算子聯(lián)系比較緊密，下游算子的故障會導(dǎo)致上游算子停止計算，陷入等待，進而使得整個流計算全部停止。Flink系統(tǒng)首先會把遠(yuǎn)程存儲的快照狀態(tài)拉取到本地，釋放之前計算任務(wù)的資源，把之前的計算任務(wù)全部取消，然后重新拉起全部計算任務(wù)，分配資源，并且把快照狀態(tài)分配給對應(yīng)的算子，將數(shù)據(jù)源消費偏移指向快照時刻的位置。因此，F(xiàn)link在開啟快照后的數(shù)據(jù)源必須具備重放的能力，以滿足故障后重新處理部分輸入數(shù)據(jù)的需求。從Flink目前的容錯機制可以明顯看出其存在的部分問題：單一節(jié)點故障導(dǎo)致整個計算圖的重啟；狀態(tài)恢復(fù)過程后數(shù)據(jù)重新計算不可避免。

因為Flink以批量的數(shù)據(jù)進行快照，相比Storm而言更加輕量，但是不可避免的是在每一次恢復(fù)時數(shù)據(jù)會有回滾計算，而且對于某些場景，快照的周期不能過短，否則會給系統(tǒng)帶來較大的負(fù)擔(dān)，影響算子的計算性能，因此這種情況下故障恢復(fù)所需要的時間也比較長，可能無法滿足場景需求。以研究的應(yīng)用場景為例，F(xiàn)link系統(tǒng)可以實現(xiàn)容錯，但是無法滿足故障后快速恢復(fù)，如果將快照周期縮短，會給系統(tǒng)帶來較大的運行負(fù)擔(dān)，影響正常的算子計算。Flink提供至少一次和精確一次的處理語義

3 Flink+系統(tǒng)設(shè)計

3.1　整體設(shè)計

本文設(shè)計的容錯方案基于快照技術(shù)進行優(yōu)化。一方面，針對大數(shù)據(jù)場景下全量狀態(tài)備份數(shù)據(jù)量巨大從而導(dǎo)致備份開銷大的問題，通過采用增量式的狀態(tài)備份（將快照間隔內(nèi)的狀態(tài)改變量定義為狀態(tài)的增量變化，即狀態(tài)增量），減少快照數(shù)據(jù)傳輸量，加快快照過程，進而縮短快照周期，從而使得故障后回滾時間變短，加快故障恢復(fù)。之所以采用狀態(tài)增量是因為大數(shù)據(jù)流計算場景下，任務(wù)的狀態(tài)數(shù)據(jù)量通常很大，而快照時間間隔內(nèi)的狀態(tài)改變量通常相對較小。如果每次快照都把全部狀態(tài)復(fù)制備份，那么相鄰兩次快照有很多冗余的備份以及網(wǎng)絡(luò)傳輸。采用增量狀態(tài)，就可以在每次快照時只對狀態(tài)改變量進行備份傳輸，在遠(yuǎn)端存儲進行全量狀態(tài)的恢復(fù)，可以大幅減小快照時的網(wǎng)絡(luò)帶寬壓力。另一方面，針對故障后系統(tǒng)在恢復(fù)過程中的時間開銷問題，通過采用備份節(jié)點以及上游輸出緩存技術(shù)，實現(xiàn)故障后任務(wù)的快速切換，并且直接消費上游緩存的輸出數(shù)據(jù)進行狀態(tài)恢復(fù)，降低故障恢復(fù)時間代價，避免單點故障向上游擴散導(dǎo)致系統(tǒng)全局回滾。

將上述容錯方案應(yīng)用到Flink系統(tǒng)，改進后的系統(tǒng)稱為Flink+，具體的設(shè)計分為以下兩個階段：

1）快照階段：采用RocksDB作為狀態(tài)后端的Flink系統(tǒng)，在增量快照時會將每個任務(wù)的快照周期內(nèi)狀態(tài)改變量發(fā)送到持久化存儲遠(yuǎn)端，基于此，可以利用持久化遠(yuǎn)端存儲的增量狀態(tài)，在備份任務(wù)上提前進行重放（Replay），保證備份任務(wù)和主任務(wù)的檢查點狀態(tài)一致性。因為傳輸?shù)目煺諣顟B(tài)主要為周期內(nèi)改變量，所以網(wǎng)絡(luò)帶寬占用會顯著下降。

2）故障恢復(fù)階段：為了加快故障恢復(fù)，在系統(tǒng)啟動時，提前在備份任務(wù)和上下游任務(wù)建立數(shù)據(jù)通路，而不是故障后再新建任務(wù)并建立通路。但該備份數(shù)據(jù)通路只在故障時工作，正常情況下不用于數(shù)據(jù)交換。故障發(fā)生時，通過備份數(shù)據(jù)通路可以做到快速的任務(wù)切換，而備份任務(wù)已有上次快照時的狀態(tài)，可直接利用上游緩存重新計算進行狀態(tài)追趕，避免計算狀態(tài)的全局回滾和恢復(fù)任務(wù)的啟動時間。

整個系統(tǒng)架構(gòu)如圖1所示，F(xiàn)link+將上層應(yīng)用的流計算轉(zhuǎn)換為對應(yīng)的工作流圖，同時選擇部分算子構(gòu)建備份流圖。在每次快照時同步主算子和備份算子之間的狀態(tài)，構(gòu)建備份算子和上游算子的數(shù)據(jù)通路。當(dāng)有備份的算子發(fā)生故障時，系統(tǒng)可以利用備份算子和提前建立的數(shù)據(jù)通路以及上游輸出緩存來快速恢復(fù)丟失的狀態(tài)。

圖 1　Flink+系統(tǒng)架構(gòu)

3.2　基于備份的快照技術(shù)

Flink系統(tǒng)的快照是基于Chandy?Lamport算法的異步改進版，通過引入barrier消息，實現(xiàn)了在不停止系統(tǒng)正常工作的情況下，完成系統(tǒng)整體狀態(tài)的備份。但是Flink系統(tǒng)在遇到節(jié)點故障后只能通過系統(tǒng)整體回滾到上一次快照狀態(tài)的方式來恢復(fù)系統(tǒng)狀態(tài)。因此可以將關(guān)鍵節(jié)點的狀態(tài)和輸入都備份起來，在故障發(fā)生時直接利用備份的信息恢復(fù)，這種做法可以控制回滾區(qū)域以及加快故障恢復(fù)的過程。具體介紹如下：

1）備份節(jié)點：本文基于Flink的快照技術(shù)，引入了備份節(jié)點的概念（在Flink中是備份任務(wù)），如圖2所示，備份節(jié)點擁有主節(jié)點的靜態(tài)資源，主要負(fù)責(zé)狀態(tài)同步，不進行計算，運行開銷較低。

圖 2　基于備份的快照

把快照的狀態(tài)增量同步給備份節(jié)點，備份節(jié)點通過合并增量狀態(tài)與上次同步的狀態(tài)，使主備份節(jié)點的狀態(tài)在每一次檢查點都是一致的；備份節(jié)點的數(shù)據(jù)通路在正常運行時處于關(guān)閉狀態(tài)，當(dāng)系統(tǒng)某個任務(wù)發(fā)生故障時，備份任務(wù)的數(shù)據(jù)通路可以立即打開，并基于同步的狀態(tài)和上游任務(wù)緩存的輸出進行恢復(fù)，從而避免系統(tǒng)整體重啟。

2）鏈?zhǔn)絺浞荩横槍浞莨?jié)點的組織形式，本文將流式處理系統(tǒng)中的有狀態(tài)任務(wù)組織成多條鏈?zhǔn)浇Y(jié)構(gòu)［25-26］，這里借鑒鏈?zhǔn)綇?fù)制的思想。假設(shè)每個任務(wù)連成一條鏈（其中為用戶指定的容錯參數(shù)，一般取=3）。成鏈的方法有多種選擇，比如考慮到故障問題，一條鏈的節(jié)點由不同機器上的節(jié)點組成，可以避免因機器故障導(dǎo)致鏈上全部節(jié)點故障；或者在考慮網(wǎng)絡(luò)帶寬的情況下，將鏈上的節(jié)點組織成為地理上直接連通的節(jié)點，減小網(wǎng)絡(luò)延遲。具體來說，可以根據(jù)任務(wù)有向無環(huán)圖（Directed Acyclic Graph，DAG）對成鏈方式進行選擇，對不同的需求適配不同的成鏈方式，成鏈方式的選擇作為本文的一個后續(xù)研究方向。

在無故障運行階段，鏈上的每一個節(jié)點（即任務(wù)）都周期性地向其后繼節(jié)點（即備份任務(wù)）同步任務(wù)狀態(tài)和計算邏輯（僅需同步一次），后繼節(jié)點作為前序節(jié)點的備份節(jié)點存在?？紤]到Flink目前的三層執(zhí)行圖模式，可以將成鏈邏輯放在Job Graph或者Execution Graph。因為任務(wù)具體執(zhí)行時分配單位是Execution，所以成鏈邏輯在Execution Graph可以更加有效地利用分配方式選擇成鏈方式。在無故障運行階段，可以利用Flink流式系統(tǒng)中流式快照的機制；不同的是Flink快照技術(shù)中任務(wù)在接收到快照消息后僅將自己的狀態(tài)寫入到持久化存儲中，而Flink+系統(tǒng)鏈?zhǔn)降貙顟B(tài)增量同步到鏈上的后繼節(jié)點直到鏈尾節(jié)點。備份任務(wù)擁有主任務(wù)的計算邏輯但無故障運行時不進行計算，只做狀態(tài)同步，為了降低內(nèi)存占用，備份任務(wù)采用RocksDB作為狀態(tài)存儲后端。

在無故障運行時，任務(wù)會在每一次快照點把狀態(tài)信息發(fā)送給備份任務(wù)，備份任務(wù)基于主任務(wù)的狀態(tài)信息維護和主任務(wù)一致檢查點狀態(tài)。備份任務(wù)同時向后繼節(jié)點發(fā)送狀態(tài)信息，使整個鏈的檢查點狀態(tài)一致。這樣，鏈上的每個節(jié)點都會擁有前序節(jié)點的同步狀態(tài)，方便故障時的恢復(fù)計算。

3）上游輸出緩存：為了避免故障恢復(fù)的時候全局回滾，本方案采用了上游備份機制。如圖3所示，在正常運行過程中，任務(wù)向下游發(fā)送自己的輸出后，并不清理本地的這些輸出數(shù)據(jù)buffer，而是緩存下來，使用空間超過內(nèi)存限制則溢出到磁盤，并在收到快照消息時記錄當(dāng)前輸出的offset，當(dāng)下游節(jié)點完成狀態(tài)同步后，會向上游發(fā)送清理輸出的消息，此時上游任務(wù)清理掉offset之前緩存的輸出數(shù)據(jù)以減少內(nèi)存占用。當(dāng)故障發(fā)生時，只將故障任務(wù)切換到備份任務(wù)，并重新消費上游備份的輸出，其他任務(wù)仍執(zhí)行之前的計算。

4）消息ID去重：考慮到故障恢復(fù)后，備份節(jié)點重新消費數(shù)據(jù)并向下游發(fā)送輸出，而故障發(fā)生之前故障節(jié)點可能已經(jīng)發(fā)送過部分相同的數(shù)據(jù)，此時下游節(jié)點則可能會處理相同的數(shù)據(jù)，從而不滿足精確一次的投遞語義。本文方案給每個消息都編上全局唯一的ID，并在每個任務(wù)中用RocksDB維護一個已處理消息ID的集合，當(dāng)檢測到消息ID在這個集合中時，則直接丟棄不進行處理，為了加快檢測速度，可以采用布隆過濾器進行過濾，當(dāng)布隆過濾器無法判斷時再訪問RocksDB進行確定。為了減少資源占用，系統(tǒng)會定期清理布隆過濾器?？紤]到對于有精確一次投遞語義需求的事務(wù)，消息ID去重是必須的，而對于較寬松的投遞語義不是必要的，且對于事務(wù)型處理來說，精確一次的投遞語義有多種實現(xiàn)方式，因此本文實驗未對消息ID進行測試。

圖 3　上游備份

總之，雖然本文方案仍然采用快照的基本機制，但是通過備份節(jié)點的方式，利用部分額外資源對主節(jié)點狀態(tài)的同步備份，可以有效節(jié)省故障后系統(tǒng)重啟計算的代價，縮短快照周期，防止故障發(fā)生時不必要的全局?jǐn)?shù)據(jù)回滾。消息去重對精確一次語義才會起作用，對于至少一次語義來說是不必要的，因此該功能可以根據(jù)上層應(yīng)用對一致性的需求開啟或者關(guān)閉。

3.3　基于任務(wù)切換的故障恢復(fù)技術(shù)

當(dāng)單點故障發(fā)生時，直接將故障任務(wù)切換到備份任務(wù)，此時備份任務(wù)啟動計算邏輯準(zhǔn)備計算，同時上游任務(wù)將緩存的輸出恢復(fù)到上一次快照的offset，并向下游發(fā)送備份的輸出數(shù)據(jù)，備份任務(wù)基于同步的狀態(tài)重新消費上游輸出數(shù)據(jù)進行狀態(tài)恢復(fù)并向下游輸出。由于僅進行任務(wù)的切換，單點故障下流式處理可以被快速恢復(fù)。

通過提前建立備份任務(wù)和上游任務(wù)以及下游任務(wù)的數(shù)據(jù)通路，在故障恢復(fù)階段，系統(tǒng)可以快速地將故障節(jié)點切換到其鏈上的后繼節(jié)點，并重新消費計算圖中上游節(jié)點的備份輸出，恢復(fù)流式計算。同時在集群中重啟故障節(jié)點任務(wù)，重啟成功后的節(jié)點可以采用追趕備份任務(wù)狀態(tài)的方式，在二者狀態(tài)同步時再次切換；或者可以把重啟后的節(jié)點作為新的備份節(jié)點添加到鏈的末尾。整個過程如圖4所示，備份算子的存在使恢復(fù)過程可以在極短的時間內(nèi)開始，而且不影響上游算子的正常工作。

圖 4　故障恢復(fù)

在Flink系統(tǒng)中，由于采用RocksDB作為狀態(tài)存儲，備份任務(wù)可能同時備份多個節(jié)點的狀態(tài)，進一步還受限于所在機器的內(nèi)存、CPU等資源，計算可能比較慢，效率相比主節(jié)點可能會比較低，有可能觸發(fā)系統(tǒng)的反壓機制，降低系統(tǒng)整體性能。因此本文方案會同時原地重啟主任務(wù)并接管原有RockDB狀態(tài)，即和備份任務(wù)同步狀態(tài)，若無法原地重啟，則在其他機器上重啟，此時該狀態(tài)為空，可以將其掛到備份任務(wù)的鏈尾。然后主任務(wù)通過其前序節(jié)點進行狀態(tài)追趕。當(dāng)完成狀態(tài)同步后，在備份任務(wù)接收到快照消息并完成狀態(tài)同步后，將重啟的主任務(wù)重新恢復(fù)成鏈頭，并將計算切換到主任務(wù)。至此，整個故障恢復(fù)完成。

4 系統(tǒng)實現(xiàn)

4.1　輕量級容錯機制

Flink原始容錯過程如圖5所示。主任務(wù)在計算過程中發(fā)生故障，任務(wù)管理器在感知到故障后會首先釋放故障任務(wù)的資源并停止整個執(zhí)行圖的計算，然后開始推導(dǎo)需要恢復(fù)的算子區(qū)域；之后，重新拉起故障區(qū)域的任務(wù)，并基于快照存儲的上一次狀態(tài)和可重放數(shù)據(jù)源進行任務(wù)的狀態(tài)恢復(fù)，系統(tǒng)重啟后重新消費自上一個快照開始的數(shù)據(jù)。

可以看出Flink原方案存在的問題在于：單個算子故障導(dǎo)致系統(tǒng)整體重啟，并且回滾后系統(tǒng)需整體重新處理自上一次快照的數(shù)據(jù)。對于簡單的流處理任務(wù)，系統(tǒng)整體重啟的代價相對較小，但是對于較大規(guī)模的流理系統(tǒng)，系統(tǒng)算子大規(guī)模重啟的代價對于實時性來說是不可接受的。

圖 5　Flink系統(tǒng)在單點故障發(fā)生后的流程

改進后Flink的容錯過程如圖6所示。本文設(shè)計的容錯機制分為狀態(tài)備份、主備份任務(wù)切換、上游輸出緩存、備份任務(wù)狀態(tài)恢復(fù)幾個模塊。

圖 6　Flink+系統(tǒng)在單點故障發(fā)生后的流程

4.2　狀態(tài)備份

Flink的快照機制會把執(zhí)行圖算子狀態(tài)保存到持久化存儲中，以便在故障時拉取用于恢復(fù)，本方案基于此，在每次快照做持久化的同時把狀態(tài)同步到備份任務(wù)，使備份任務(wù)維持和主任務(wù)一樣的快照狀態(tài)。為了方便實現(xiàn)，改進措施直接利用Flink原有的狀態(tài)恢復(fù)過程，在快照時，備份節(jié)點進行狀態(tài)“恢復(fù)”，即同步狀態(tài)。在成鏈方式上，因為備份節(jié)點和主節(jié)點之間的聯(lián)系屬于備份層面的，并不是實際流處理的數(shù)據(jù)通路，因此成鏈作為在Flink三層圖結(jié)構(gòu)中Execution的一個單獨抽象來對待，通過記錄節(jié)點之間的成鏈關(guān)系，在快照時同步狀態(tài)，在故障時切換，把上游節(jié)點的輸出切換到備份節(jié)點的通路上。同時，因為使用了RocksDB作為狀態(tài)后端，在快照時，可以利用RocksDB本身的changelog實現(xiàn)增量式的狀態(tài)備份，即：在每次快照時只發(fā)送新增的或者壓縮的狀態(tài)文件，未改變的狀態(tài)文件不再進行傳輸。增量形式使快照傳輸?shù)木W(wǎng)絡(luò)流量大幅降低，后續(xù)實驗結(jié)果也體現(xiàn)了這一點。

4.3　主備份任務(wù)切換

備份任務(wù)在正常情況下啟動后就直接掛起，只在狀態(tài)同步時工作，不觸發(fā)任何計算邏輯，不占用CPU資源。當(dāng)主任務(wù)發(fā)生故障時，系統(tǒng)檢測到后會立即把主任務(wù)掛起，把數(shù)據(jù)通路切換到備份任務(wù)上，并且啟動備份計算邏輯。這種情況下，備份節(jié)點時刻處于熱啟動狀態(tài)，但是幾乎不占用CPU資的計算資源。

在Flink系統(tǒng)中，正常的流處理執(zhí)行圖中的上下游任務(wù)會通過數(shù)據(jù)通路channel實現(xiàn)數(shù)據(jù)交換，且上下游任務(wù)之間的channel是一一對應(yīng)關(guān)系；發(fā)生故障后，備份任務(wù)可以通過重新建立和上游任務(wù)以及下游任務(wù)的數(shù)據(jù)通路來代替主任務(wù)執(zhí)行計算，但是這一過程需要花費一定的時間，且隨著計算圖規(guī)模變大，花費時間越多。因此，為了實現(xiàn)快速的任務(wù)切換，本文的方案會讓備份任務(wù)和主任務(wù)一樣，提前把數(shù)據(jù)通路建立好并通過標(biāo)記來控制數(shù)據(jù)的交換。

Flink的數(shù)據(jù)通路在向下游發(fā)送數(shù)據(jù)時，可以通過添加標(biāo)記來控制數(shù)據(jù)是否被實際發(fā)送，改進后的系統(tǒng)同樣通過一個flag實現(xiàn)數(shù)據(jù)通路的控制，在正常情況下，該flag值為true，會使得上游算子的數(shù)據(jù)只發(fā)送給主任務(wù)，發(fā)生故障后，會將flag置為false，使數(shù)據(jù)可以被備份任務(wù)的數(shù)據(jù)通路接受。

4.4　上游輸出緩存

Flink的每個任務(wù)會把輸出緩存在buffer中，由管理器通知下游算子buffer數(shù)據(jù)可以被消費，本文基于Flink的buffer，將其緩存下來，在每個快照周期清理一次，僅保存上一次快照期間的輸出數(shù)據(jù)，以滿足故障恢復(fù)的需求。對于快照時的offset，F(xiàn)link會一起保存到每一次快照中，因此在故障后恢復(fù)狀態(tài)時，offset會自動指向上一次快照時的輸出位置；改進措施則是在備份節(jié)點中同步記錄該offset值，在故障時直接使用備份節(jié)點的offset。如果buffer使用的內(nèi)存超過了限額，F(xiàn)link會使用磁盤來緩存。有了上游輸出備份，在故障后下游算子可以直接消費上游算子緩存的數(shù)據(jù)恢復(fù)狀態(tài)，而不需要從頭開始消費數(shù)據(jù)，很大程度上減弱了故障的影響。Flink系統(tǒng)本身提供了從持久化存儲中拉取到本地的狀態(tài)文件來恢復(fù)狀態(tài)的API，同時，因為Flink的不同模塊的通信由Akka提供，兩個任務(wù)之間沒有直接通信方式，均通過執(zhí)行管理器來調(diào)度；因此為了方便實現(xiàn)狀態(tài)同步，在系統(tǒng)設(shè)計中，基于上述API進行改進，結(jié)合快照協(xié)調(diào)器的消息，當(dāng)快照結(jié)束后，備份任務(wù)收到執(zhí)行管理器的消息便開始從持久化遠(yuǎn)端拉取對應(yīng)的狀態(tài)文件并在本地恢復(fù)。這種方式比主任務(wù)直接把快照狀態(tài)文件傳輸給備份任務(wù)要花費更多的時間，但是實現(xiàn)更加簡單。

4.5　備份任務(wù)狀態(tài)恢復(fù)

故障后備份任務(wù)基于上游緩存的數(shù)據(jù)和之前同步的主任務(wù)狀態(tài)來進行狀態(tài)恢復(fù)，但是此時上游算子的計算并沒有被停止，系統(tǒng)整體仍處于運行狀態(tài)，故障算子則在恢復(fù)后從故障時間點開始重新處理上游緩存的數(shù)據(jù)，追趕備份任務(wù)的狀態(tài)?？紤]到這一點，系統(tǒng)可能會因為故障算子的重復(fù)處理產(chǎn)生反壓問題，但考慮到故障能夠被很快恢復(fù)，反壓問題可能并不嚴(yán)重。

Flink+在面臨算子子任務(wù)崩潰時的恢復(fù)過程整體變得較為復(fù)雜，但是避免了整個執(zhí)行圖的重啟，只對故障的算子進行重啟，同時備份算子可以立即切換過來執(zhí)行計算任務(wù)，加快了恢復(fù)過程

5 實驗與結(jié)果分析

為了驗證本文提出的輕量級故障恢復(fù)方案，在單機模式和分布式環(huán)境下分別對Flink系統(tǒng)（1.13.0版本）和改進后的Flink+系統(tǒng)進行測試。實驗采用WordCount任務(wù)來測試流式系統(tǒng)的計算、備份、故障恢復(fù)能力，數(shù)據(jù)源由英文版《哈利波特》構(gòu)成。

1）實驗?zāi)康?。針對流式系統(tǒng)備份開銷問題和故障恢復(fù)延遲問題，本文設(shè)計一種基于增量狀態(tài)備份的快照容錯方案，并在Apache Flink系統(tǒng)上實現(xiàn)了原型，本實驗的目的是為了驗證該方案的可行性和有效性，探究增量狀態(tài)對快照速度的影響以及上游輸出備份和狀態(tài)備份對故障恢復(fù)速度的影響。

2）實驗內(nèi)容。利用WordCount流式計算任務(wù)，對Flink+系統(tǒng)的故障容錯能力進行驗證。實驗對單機和分布式集群下的Flink系統(tǒng)和Flink+系統(tǒng)進行對比探究，基于不同任務(wù)并行度，測試了Flink和Flink+的故障恢復(fù)速度。為了驗證改進部件對Flink系統(tǒng)本身的影響可以忽略不計，實驗也對同一任務(wù)下兩個系統(tǒng)的CPU占用率和內(nèi)存占用率進行了測試。

3）單點故障。Flink中最小運行單位是Execution任務(wù)，會被分配給一個Java虛擬機（Java Virtual Machine， JVM）線程執(zhí)行，本文將其在運算過程發(fā)生異常導(dǎo)致自身崩潰的問題定義為單點故障。

4）機器故障。單個機器上可能運行多個JVM，每個JVM可以運行多個執(zhí)行管理器（TaskManager），每個執(zhí)行管理器可以調(diào)度運行多個Execution任務(wù)。機器的故障會導(dǎo)致多個JVM的故障，進而導(dǎo)致多個單點故障。機器故障可以具化為多個單點故障，因此機器故障導(dǎo)致的多個單點故障如果相互無關(guān)，那么可以被視為多個單點故障的處理；如果多個單點故障有關(guān)系，可以把有聯(lián)系的單點故障認(rèn)為是一個統(tǒng)一的大的單點故障，其恢復(fù)流程和普通的單點故障基本一致。因此實驗過程只針對單點故障。

5.1　實驗環(huán)境

實驗用到的物理機均為16核32線程的Linux機器，CPU為Intel Xeon E5?2650，主頻為2.00 GHz，每臺機器的內(nèi)存為256 GB，操作系統(tǒng)均為Ubuntu 16.04.10。

Flink系統(tǒng)的部署主要分為Taskmanager和Jobmanager兩個部件：Jobmanager負(fù)責(zé)整個流式任務(wù)的調(diào)度執(zhí)行；Taskmanager負(fù)責(zé)最小粒度單位“并行任務(wù)”的執(zhí)行，其內(nèi)部有slot，是用戶代碼實際執(zhí)行的地方。

單機實驗?zāi)Ｊ较?，實驗環(huán)境由1臺16核32線程的機器構(gòu)成，F(xiàn)link組件由1個Jobmanager和16個Taskmanager構(gòu)成，每個Taskmanager僅包含一個slot執(zhí)行單位。

分布式集群實驗?zāi)Ｊ较?，實驗環(huán)境由8臺16核32線程的機器構(gòu)成，F(xiàn)link組件由1個Jobmanager（部署在控制機器上1上）和16個Taskmanager（每臺機器部署2個Taskmanager）構(gòu)成，同樣的，每個Taskmanager僅由一個slot構(gòu)成。

這里每個Taskmanager僅分配一個slot是為了使任務(wù)之間通過網(wǎng)絡(luò)通信而不是Taskmanager內(nèi)部的內(nèi)存空間通信，盡可能模擬真實大數(shù)據(jù)場景下的網(wǎng)絡(luò)通信場景。

5.2　實驗結(jié)果

從表1的實驗結(jié)果可以看出，基于本文設(shè)計改進后的Flink+系統(tǒng)在單機模式下的整體表現(xiàn)都明顯優(yōu)于原Flink系統(tǒng)，其在系統(tǒng)故障后的恢復(fù)時間可以達(dá)到數(shù)十毫秒到數(shù)百毫秒級，而Flink系統(tǒng)面對單點故障需要注銷原有資源，重新拉起任務(wù)，恢復(fù)時間在秒級。

表 1　單機模式下Flink和Flink+系統(tǒng)恢復(fù)時間對比

為了探究并行度對故障恢復(fù)的影響，本文針對不同的并行度進行了實驗測試。實驗發(fā)現(xiàn)，對于同一任務(wù)的不同并行度，隨著任務(wù)并行度的增大，恢復(fù)時間減小比例基本上在逐漸增大，說明改進后的系統(tǒng)在高并行度下表現(xiàn)更加良好，相較于原Flink系統(tǒng)，更能適應(yīng)并行化任務(wù)。觀察實驗結(jié)果還能發(fā)現(xiàn)隨著并行度的增加，兩個系統(tǒng)的故障恢復(fù)時間都呈現(xiàn)先減小后增大的趨勢。

造成這一現(xiàn)象的主要原因：單機下資源總數(shù)受限，一方面，提高并行度可以充分利用CPU的并行能力加快故障恢復(fù)，縮短恢復(fù)時間；而另一方面，并行任務(wù)增多意味著CPU負(fù)荷變大，一定程度上會延長恢復(fù)時間。當(dāng)并行度逐漸接近CPU核數(shù)時，提高并行度帶來的增益被多任務(wù)的負(fù)荷抵消，甚至產(chǎn)生負(fù)增益，導(dǎo)致故障恢復(fù)時間增加。因此，可以看出流式任務(wù)的并行度并不是越大越好，適中的并行度有利于故障后的恢復(fù)。

考慮到實際環(huán)境中流計算任務(wù)往往是分布式進行，本文對分布式環(huán)境下的Flink+系統(tǒng)和Flink系統(tǒng)的故障恢復(fù)表現(xiàn)進行了實驗測試。從表2的實驗結(jié)果可以看出，改進后的Flink+系統(tǒng)在分布式環(huán)境下同樣表現(xiàn)優(yōu)異，基本能夠維持在200 ms以內(nèi)的故障恢復(fù)時間。但是因為分布式環(huán)境下不可避免的網(wǎng)絡(luò)通信延遲，系統(tǒng)在高并行度任務(wù)上并沒有單機模式那樣優(yōu)異的表現(xiàn)，本研究認(rèn)為這主要是網(wǎng)絡(luò)延遲帶來的影響。

表 2　分布式環(huán)境下Flink和Flink+系統(tǒng)恢復(fù)時間對比

因為單機模式和分布式集群的區(qū)別主要在于Taskmanager的部署位置不同，二者對于資源的使用大致相同，而實驗中給每個Taskmanager僅配置了1個slot資源，因此在分布式環(huán)境下當(dāng)并行度達(dá)到32時，并行任務(wù)之間需要共享slot資源，此時，并行任務(wù)帶來的負(fù)增益大于并行計算帶來的正增益，因此整體恢復(fù)時間變長，正如實驗結(jié)果顯示，在并行度為32時，F(xiàn)link系統(tǒng)和改進后的Flink+系統(tǒng)表現(xiàn)均有所下降。

為了驗證本文的容錯機制對于流式系統(tǒng)本身并沒有較大的影響，即可以在滿足輕量性的同時加快系統(tǒng)在故障時的恢復(fù)，本文針對Flink系統(tǒng)和改進后的Flink+系統(tǒng)進行了CPU和內(nèi)存測試實驗。

實驗由2臺16核32線程的機器構(gòu)成的集群進行，機器1負(fù)責(zé)kafka數(shù)據(jù)讀入和Jobmanager執(zhí)行，機器2負(fù)責(zé)Taskmanager任務(wù)執(zhí)行。通過對機器2上運行的Taskmanager的檢測，實驗獲取了執(zhí)行同樣任務(wù)處理同樣數(shù)據(jù)情況下兩個系統(tǒng)的CPU和內(nèi)存占用率結(jié)果，為了方便獲取數(shù)據(jù)，本實驗在較小規(guī)模下進行，僅在機器2上開啟了兩個Taskmanager進行測試。

如圖7所示，F(xiàn)link和Flink+的整體執(zhí)行過程幾乎完美吻合，二者在任務(wù)過程中占用率幾乎相等，CPU占用率維持在200%～300%，這說明改進后的Flink+并沒有給系統(tǒng)帶來較大的CPU負(fù)擔(dān)。通過計算兩個系統(tǒng)的平均CPU占用率可得：Flink系統(tǒng)的平均CPU占用率為264%，F(xiàn)link+系統(tǒng)的平均占用率是258.8%，考慮到其他進程以及CPU占用率的波動，可以認(rèn)為二者占用率基本一致，即本文的改進沒有給Flink系統(tǒng)帶來較大的負(fù)擔(dān)。

在內(nèi)存方面，二者的占用率也維持在相同的水平，實驗結(jié)果顯示并無差距。

綜合實驗結(jié)果，可以得出如下結(jié)論：本文提出的輕量級快照容錯方案可以大幅減少故障恢復(fù)時間；流式任務(wù)的并行度太高會導(dǎo)致故障后恢復(fù)時間增加；本文提出的方案不會給流式系統(tǒng)本身帶來顯著CPU壓力和內(nèi)存壓力。

圖 7　Flink與Flink+的CPU占用率對比

6 結(jié)語

本文針對流式系統(tǒng)分布式快照機制故障恢復(fù)慢的問題，提出了一種基于增量狀態(tài)備份的故障恢復(fù)方法Flink+，通過增量式狀態(tài)同步實現(xiàn)了快速狀態(tài)備份，通過上游輸出緩存與提前網(wǎng)絡(luò)連接進一步細(xì)化了故障恢復(fù)粒度并實現(xiàn)了快速故障恢復(fù)。實現(xiàn)結(jié)果表明，F(xiàn)link+能夠在不顯著增加額外CPU、內(nèi)存開銷的同時實現(xiàn)6～8倍的故障恢復(fù)加速。

[1] ZHANG Y Y， LI J X， ZHANG Y M， et al. FreeLauncher： lossless failure recovery of parameter servers with ultralight replication［C］// Proceedings of the IEEE 41st International Conference on Distributed Computing Systems. Piscataway： IEEE， 2021： 472-482.

[2] ZHANG Y Y， LI J X， SUN C G， et al. HotML： a DSM?based machine learning system for social networks［J］. Journal of Computational Science， 2018， 26： 478-487.

[3] CARBONE P， KATSIFODIMOS A， EWEN S， et al. Apache Flink： stream and batch processing in a single engine［J］. Bulletin of the IEEE Computer Society Technical Committee on Data Engineering， 2015， 38（4）： 28-38.

[4] CARBONE P， EWEN S， FóRA G， et al. State management in Apache Flink： consistent stateful distributed stream processing［J］. Proceedings of the VLDB Endowment， 2017， 10（12）： 1718-1729.

[5] GARCíA?GIL D， RAMíREZ?GALLEGO S， GARCíA S， et al. A comparison on scalability for batch big data processing on Apache Spark and Apache Flink［J］. Big Data Analytics， 2017， 2： No.1.

[6] MENG X R， BRADLEY J， YAVUZ B， et al. MLlib： machine learning in Apache Spark［J］. Journal of Machine Learning Research， 2016， 17： 1-7.

[7] ZAHARIA M， DAS T， LI H Y， et al. Discretized streams： fault?tolerant streaming computation at scale［C］// Proceedings of the 24th ACM Symposium on Operating Systems Principles. New York： ACM， 2013： 423-438.

[8] ZAHARIA M， CHOWDHURY M， DAS T， et al. Resilient distributed datasets： a fault?tolerant abstraction for in?memory cluster computing［C］// Proceedings of the 9th USENIX Symposium on Networked Systems Design and Implementation. Berkeley： USENIX Association， 2012： 15-28.

[9] ZAHARIA M， DAS T， LI H Y， et al. Discretized streams： an efficient and fault?tolerant model for stream processing on large clusters［C］// Proceedings of the 4th USENIX Workshop on Hot Topics in Cloud Computing. Berkeley： USENIX Association， 2012： No.10.

[10] TOSHNIWAL A， TANEJA S， SHUKLA A， et al. Storm@Twitter［C］// Proceedings of the 2014 ACM SIGMOD international Conference on Management of Data. New York： ACM， 2014： 147-156.

[11] IQBAL M H， SOOMRO T R. Big data analysis： Apache Storm perspective［J］. International Journal of Computer Trends and Technology， 2015， 19（1）： 9-14.

[12] NOGHABI S A， PARAMASIVAM K， PAN Y， et al. Samza： stateful scalable stream processing at LinkedIn［J］. Proceedings of the VLDB Endowment， 2017， 10（12）： 1634-1645.

[13] CHANDY K M， LAMPORT L. Distributed snapshots： Determining global states of distributed systems［J］. ACM Transactions on Computer Systems， 1985， 3（1）： 63-75.

[14] CARBONE P， FóRA G， EWEN S， et al. Lightweight asynchronous snapshots for distributed dataflows［EB/OL］. （2015-06-29）［2021-12-15］.https：//arxiv.org/pdf/1506.08603.pdf.

[15] 段澤源. 大數(shù)據(jù)流式處理系統(tǒng)負(fù)載均衡與容錯機制的研究［D］. 北京：華北電力大學(xué)， 2017： 28-30.（DUAN Z Y. Research on load balancing and fault tolerant mechanism of big data stream processing system［D］. Beijing： North China Electric Power University， 2017：28-30.）

[16] 孫大為，張廣艷，鄭緯民. 大數(shù)據(jù)流式計算：關(guān)鍵技術(shù)及系統(tǒng)實例［J］. 軟件學(xué)報， 2014， 25（4）：839-862.（SUN D W， ZHANG G Y， ZHENG W M. Big data stream computing： technologies and instances［J］. Journal of Software， 2014， 25（4）： 839-862.）

[17] LI H L， WU J， JIANG Z， et al. Integrated recovery and task allocation for stream processing［C］// Proceedings of the IEEE 36th International Performance Computing and Communications Conference. Piscataway： IEEE， 2017： 1-8.

[18] LI H L， WU J， JIANG Z， et al. Task allocation for stream processing with recovery latency guarantee［C］// Proceedings of the 2017 IEEE International Conference on Cluster Computing. Piscataway： IEEE， 2017： 379-383.

[19] AKIDAU T， BALIKOV A， BEKIRO?LU K， et al. MillWheel： fault?tolerant stream processing at Internet scale［J］. Proceedings of the VLDB Endowment， 2013， 6（11）： 1033-1044.

[20] GUO J， AGRAWAL G. Smart Streaming： a high?throughput fault? tolerant online processing system［C］// Proceedings of the 2020 IEEE International Parallel and Distributed Processing Symposium Workshops. Piscataway： IEEE， 2020： 396?405.

[21] LIN C F， ZHAN J J， CHEN H H， et al. Ares： a high performance and fault?tolerant distributed stream processing system［C］// Proceedings of the IEEE 26th International Conference on Network Protocols. Piscataway： IEEE， 2018： 176-186.

[22] VENKATARAMAN S， PANDA A， OUSTERHOUT K， et al. Drizzle： fast and adaptable stream processing at scale［C］// Proceedings of the 26th Symposium on Operating Systems Principles. New York： ACM， 2017： 374-389.

[23] LIU P C， XU H L， DA SILVA D， et al. FP4S： fragment?based parallel state recovery for stateful stream applications［C］// Proceedings of the 2020 IEEE International Parallel and Distributed Processing Symposium. Piscataway： IEEE， 2020： 1102-1111.

[24] XU H L， LIU P C， CRUZ?DIAZ S， et al. SR3： customizable recovery for stateful stream processing systems［C］// Proceedings of the 21st International Middleware Conference. New York： ACM， 2020： 251-264.

[25] RENESSE R van， SCHNEIDER F B. Chain replication for supporting high throughput and availability［C］// Proceedings of the 6th Symposium on Operating Systems Design and Implementation. Berkeley： USENIX Association， 2004： 91-104.

[26] TERRACE J， FREEDMAN M J. Object storage on CRAQ： high? throughput chain replication for read?mostly workloads［C］// Proceedings of the 2009 USENIX Annual Technical Conference. Berkeley： USENIX Association， 2009： No.11.

Efficient failure recovery method for stream data processing system

LIU Yang1，2，3， ZHANG Yangyang1，2， ZHOU Haoyi1，2，4*

（1，，100191，；2，，100191，；3，，100191，；4，，100191，）

Focusing on the issue that the single point of failure cannot be efficiently handled by streaming data processing system Flink， a new fault?tolerant system based on incremental state and backup， Flink+， was proposed. Firstly， backup operators and data paths were established in advance. Secondly， the output data in the data flow diagram was cached， and disks were used if necessary. Thirdly， task state synchronization was performed during system snapshots. Finally， backup tasks and cached data were used to recover calculation in case of system failure. In the system experiment and test， Flink+ dose not significantly increase the additional fault tolerance overhead during fault?free operation； when dealing with the single point of failure in both single?machine and distributed environments， compared with Flink system， the proposed system has the failure recovery time reduced by 96.98% in single?machine 8?task parallelism and by 88.75% in distributed 16?task parallelism. Experimental results show that using incremental state and backup method together can effectively reduce the recovery time of the single point of failure of the stream system and enhance the robustness of the system.

stream data processing system; failure recovery; distributed checkpoint; state backup; Apache Flink

This work is partially supported by National Natural Science Foundation of China （U20B2053， 61872022）， Open Project of State Key Laboratory of Software Development Environment （SKLSDE?2020ZX?12）.

LIU Yang， born in 1999， Ph. D. candidate. His research interests include distributed systems， graph processing systems.

ZHANG Yangyang， born in 1991， Ph. D. candidate. His research interests include distributed systems， machine learning， graph processing.

ZHOU Haoyi， born in 1991， Ph. D.， lecturer. His research interests include big data system， machine learning.

1001-9081（2022）11-3337-09

10.11772/j.issn.1001-9081.2021122108

2021?12?15；

2022?02?27；

2022?03?04。

國家自然科學(xué)基金資助項目（U20B2053， 61872022）；軟件開發(fā)環(huán)境國家重點實驗室開放課題（SKLSDE?2020ZX?12）。

TP311.5

A

劉陽（1999—），男，山西大同人，博士研究生，CCF會員，主要研究方向：分布式系統(tǒng)、圖計算系統(tǒng)；張揚揚（1991—），男，河北保定人，博士研究生，CCF會員，主要研究方向：分布式系統(tǒng)、機器學(xué)習(xí)、圖計算；周號益（1991—），男，四川德陽人，講師，博士，CCF會員，主要研究方向：大數(shù)據(jù)系統(tǒng)、機器學(xué)習(xí)。