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

劉陽(yáng)，張揚(yáng)揚(yáng)，周號(hào)益

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

劉陽(yáng)1，2，3，張揚(yáng)揚(yáng)1，2，周號(hào)益1，2，4*

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

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

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

0 引言

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

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

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

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

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

本文的主要工作如下：

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

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

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

1 研究背景

1.1　語(yǔ)義支持

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

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

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

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

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

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

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

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

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

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

2 Flink

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

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

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

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

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

2.2　Flink分布式快照

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

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

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

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

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

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

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

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

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

3.1　整體設(shè)計(jì)

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

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

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

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

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

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

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

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

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

圖 2　基于備份的快照

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

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

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

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

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

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

圖 3　上游備份

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

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

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

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

圖 4　故障恢復(fù)

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

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

4.1　輕量級(jí)容錯(cuò)機(jī)制

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

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

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

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

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

4.2　狀態(tài)備份

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

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

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

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

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

4.4　上游輸出緩存

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

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

故障后備份任務(wù)基于上游緩存的數(shù)據(jù)和之前同步的主任務(wù)狀態(tài)來(lái)進(jìn)行狀態(tài)恢復(fù)，但是此時(shí)上游算子的計(jì)算并沒(méi)有被停止，系統(tǒng)整體仍處于運(yùn)行狀態(tài)，故障算子則在恢復(fù)后從故障時(shí)間點(diǎn)開(kāi)始重新處理上游緩存的數(shù)據(jù)，追趕備份任務(wù)的狀態(tài)。考慮到這一點(diǎn)，系統(tǒng)可能會(huì)因?yàn)楣收纤阕拥闹貜?fù)處理產(chǎn)生反壓?jiǎn)栴}，但考慮到故障能夠被很快恢復(fù)，反壓?jiǎn)栴}可能并不嚴(yán)重。

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

5 實(shí)驗(yàn)與結(jié)果分析

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

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

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

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

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

5.1　實(shí)驗(yàn)環(huán)境

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

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

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

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

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

5.2　實(shí)驗(yàn)結(jié)果

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

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

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

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

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

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

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

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

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

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

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

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

圖 7　Flink與Flink+的CPU占用率對(duì)比

6 結(jié)語(yǔ)

本文針對(duì)流式系統(tǒng)分布式快照機(jī)制故障恢復(fù)慢的問(wèn)題，提出了一種基于增量狀態(tài)備份的故障恢復(fù)方法Flink+，通過(guò)增量式狀態(tài)同步實(shí)現(xiàn)了快速狀態(tài)備份，通過(guò)上游輸出緩存與提前網(wǎng)絡(luò)連接進(jìn)一步細(xì)化了故障恢復(fù)粒度并實(shí)現(xiàn)了快速故障恢復(fù)。實(shí)現(xiàn)結(jié)果表明，F(xiàn)link+能夠在不顯著增加額外CPU、內(nèi)存開(kāi)銷(xiāo)的同時(shí)實(shí)現(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ù)載均衡與容錯(cuò)機(jī)制的研究［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ù)流式計(jì)算：關(guān)鍵技術(shù)及系統(tǒng)實(shí)例［J］. 軟件學(xué)報(bào)， 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。

國(guó)家自然科學(xué)基金資助項(xiàng)目（U20B2053， 61872022）；軟件開(kāi)發(fā)環(huán)境國(guó)家重點(diǎn)實(shí)驗(yàn)室開(kāi)放課題（SKLSDE?2020ZX?12）。

TP311.5

A

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