MapReduce 框架下結合分布式編碼計算的容錯算法

張 基，謝在鵬，毛鶯池，徐媛媛，朱曉瑞，李博文

（河海大學計算機與信息學院，南京 211100）

0 概述

容錯技術是分布式系統的重要組成部分，確保了發生故障時的系統連續性和功能性［1］。近年來，隨著分布式系統規模的不斷擴大、分布式架構和計算復雜度的日益增加［2］以及廉價商業硬件的廣泛使用，使得計算任務發生故障的概率持續增加，例如在Google 生產集群中平均每天會有數十個節點發生故障［3］。瞬態故障尤其是軟錯誤［4］會導致計算機系統的異常行為，這類故障會損壞數據的完整性，引起分布式節點的計算失效［4-5］。在一些大型分布式系統中平均每天會有1%～2%的節點發生失效［6］，因此容錯技術可在保證部分節點失效的情況下，使分布式系統仍能繼續運行且得到正確結果［7-9］。在基于MapReduce［10］的分布式計算中，數據洗牌（shuffle）階段較高的通信開銷嚴重影響了分布式計算性能，例如在Facebook 的Hadoop 集群中，33%的任務執行總時間用于數據洗牌階段［11］。針對基于MapReduce 分布式計算框架的多副本容錯算法通信開銷較大的問題，本文提出一種結合分布式編碼計算的容錯算法CTMR，使基于MapReduce 的分布式計算系統在發生瞬態故障的情況下仍能繼續運行且得到正確結果，同時能有效降低容錯計算過程中的通信開銷。

1 相關工作

基于多副本冗余技術［12-15］的分布式容錯算法于1962年由IBM提出，但在現代分布式系統中仍然被廣泛采用［16-17］。這類算法的主要思想是：在含有n個節點的分布式系統中，如果每個節點要容忍f個故障，那么該節點可以使用（f+1）個獨立的副本，顯然存儲和運行這些副本需要消耗大量的空間和其他資源。此外，由于多副本之間的一致性使得基于副本冗余技術的容錯算法更易于設計，同時副本的維護和恢復成本較低。三模冗余（Triple Modular Redundancy，TMR）是軟件和硬件系統中常用的基于副本的容錯算法［18-19］，使用3 個實現功能相同的模塊同時進行操作，系統使用投票機制選出最終輸出。由于3 個模塊相互獨立，同時有2 個模塊出現相同錯誤的概率非常小，因此可大幅提高系統可靠性，掩蔽故障模塊的錯誤。文獻［20］提出一種優化的副本虛擬機放置算法，動態設置k個副本虛擬機并將其作為備份來提高云服務的可靠性。在基于MapReduce的分布式計算［10］中，每個任務被分配到多個節點以確保在某個節點出現故障的情況下系統仍能繼續運行。文獻［21］提出面向分布式流體系結構的多副本容錯技術，該技術通過比較多個副本的數據進行檢錯，采取三取二的邏輯判斷方式選擇多數相同的結果，并對錯誤的數據進行糾錯以防止錯誤進一步傳播，但當數據量較大時通信開銷會成為容錯過程中的性能瓶頸。文獻［18］提出一種兩階段三模冗余（two-stage TMR）容錯算法，該算法將每個任務備份3 份并分配給3 個節點，先指定2 個節點進行計算，當2 個節點執行完畢后進行結果比較，如果比較結果不一致，則指定第3 個節點進行計算，最終通過投票選出多數一致的結果。該算法在故障率較低的情況下可以有效節省系統資源，但當錯誤率較高時，重新執行第3 份備份任務不僅增加了計算工作量并且大幅降低了系統實時性能。

通信開銷是基于MapReduce 的分布式計算中的主要性能瓶頸，這是因為在數據洗牌階段交換大量中間結果。為解決該問題，文獻［7］提出分布式編碼計算方法，將map 任務重復布置到多個不同的節點，通過增加計算冗余創建同時滿足多個服務器數據需求的編碼數據來降低通信開銷。在一個分布式集群中，node1包含數據集{v1，v2}，node2包含數據集{v2，v3}，node3包含數據集{v1，v3}。node1將本地數據集編碼結果c1（c1=v1⊕v2，⊕表示異或）運用廣播發出。node2接收到編碼數據后利用本地數據v2即可將收到的編碼結果解碼得到v1，其中v1=v2⊕c1。同理，node3通過解碼即可得到v2。在該過程中，通過將node1與node2和node3冗余存儲的數據{v1，v2}進行編碼來創建同時滿足node2和node3的編碼數據，而無需將v1、v2分別發送至node2和node3，從而降低通信開銷。然而，目前尚未發現能有效降低分布式容錯計算中通信開銷的相關討論和研究。為降低現有分布式計算使用多副本容錯過程中產生的通信開銷，本文基于副本冗余和分布式編碼計算技術，提出一種CTMR 容錯算法。

2 CTMR 容錯算法

2.1 CTMR 模型

假設在分布式集群中有N個節點，標記為{node1，node2，…，noden}，N個節點協作完成一個計算任務。當前集群待處理的數據量為M，每份數據的冗余度為r，將該數據集分配到當前集群的N個map節點，每個節點的數據量為M?。map 節點對數據進行計算，將產生的中間結果使用分布式編碼計算得到的編碼數據發送到Q個reduce 節點，其中編碼壓縮比為c，即將c個中間結果通過編碼得到一個編碼中間結果。因此，每個reduce 節點將收到其余（N?Q）個map 節點的編碼中間結果，如式（1）所示：

reduce 節點對接收到的編碼中間結果進行解碼，通過校驗識別發生故障的編碼數據包，并利用冗余中間結果得到M份數據最終正確的計算結果，如式（2）所示：

因為本文算法隨機選取Q個reduce 節點均能夠完成所有中間結果的驗證，所以節點應該包含編碼數據所有可能的中間結果，如式（3）所示：

由于TMR 的低復雜度及高可靠性，因此本文選取r=3、c=2，聯立式（1）～式（3）解得N=6、M=8，即在含有6 個節點的分布式集群中，將待處理的數據分成8 個獨立的子數據塊，每個數據塊冗余3 份，每個節點包含4 份數據，使用故障檢測與恢復算法進行容錯計算，同時降低計算過程中的通信開銷。

在包含N個節點的分布式集群中，將每6 個節點分為1 個子集群，共有組子集群。各個子集群之間相互獨立，而每個子集群內的節點在邏輯上是相鄰的，在物理上可以是分散的。將計算任務分配到個子集群上，各個子集群并行計算。每個子集群的數據量為，同時將子集群的數據集劃分為8 個獨立的子數據塊，每個數據塊復制3 份。每個子集群使用如圖1 所示的CTMR 模型進行表示，其中立方體的每個面表示1 個節點，每個面的4 個頂點代表當前節點包含的4 個數據集。立方體中的每條棱代表與該棱鄰接的兩個面所表示的節點公共冗余數據集，而每個面所表示的節點數據集用Bi表示，例如node1本地數據集為B1={b1，b2，b3，b4}。在模型中每個頂點與3 個面鄰接，即每份數據冗余3 份并存放至3 個節點上，例如b3分別存放至node1、node3和node5上。nodek在map 階段對本地數據集的每一個數據塊bi執行map 函數F(bi)得到相應的中間結果集。

圖1 CTMR 模型Fig.1 CTMR model

兩個鄰近節點nodek和nodem冗余存儲的數據集用Rk，m表示，例如R1，3={b1，b3}。將nodek與nodem冗余存儲的數據Rk，m所對應的中間結果使用分布式編碼計算得到編碼結果uk，m，如式（4）所示：

其中，⊕表示異或操作。nodek在reduce 階段的函數為H(v1，v2，…，vi)=rk。

2.2 CTMR 故障檢測與恢復

將一個包含N個節點的分布式集群劃分為個子集群，各個子集群并行計算的同時進行檢錯和糾錯。每個子集群中的6 個節點分別表示為nodek、nodem、nodes、noden、nodep、nodeq，其中nodek和nodem、nodes和noden、nodep和nodeq分別為CTMR模型中的3 組對面?？梢钥闯?，任意一組對面中的兩個節點的本地數據集的交集為空，但并集是當前子集群數據集的全集。每個節點map 階段通過函數F(bi)計算本地數據集得到相應的中間結果集，例如nodek通過計算本地數據集Bk得到相應的中間結果集。

隨機選取一個節點nodek，將其在CTMR 模型中的對面節點nodem作為校驗節點。nodes、noden、nodep、nodeq為與nodek相鄰的節點，將其與nodek冗余存儲的數據集Rs，k、Rn，k、Rp，k、Rq，k所對應的中間結果進行編碼，得到相應的編碼結果us，k、un，k、up，k、uq，k。將對應的編碼結果發送給nodek，然后在nodem上執行相同過程。數據塊分發與中間結果編碼的偽代碼如算法1 所示。

算法1數據分發與中間結果編碼算法

nodek和nodem通過比較接收到的編碼數據包與當前節點產生的中間結果來驗證數據的正確性。若式（5）兩個等式中的任意一個成立，則表明nodek本地數據集的運算結果正確，通過reduce 函數可得到當前節點運算結果。

若式（5）中兩個等式均不成立，而式（6）兩個等式中的任意一個成立，則假設us，k驗證成功，即us，k=

若式（7）成立，則表明nodek本地有數據計算錯誤，但是收到的其他節點的編碼數據包正確。此時，通過reduce 函數得到當前節點數據集的正確結果為

若us，k、un，k、up，k、uq，k全部驗證失敗，則重新選取兩個校驗節點進行上述操作。當nodek與nodem均能通過驗證得到正確運算結果時，利用reduce 函數便可得到CTMR 模型的最終結果r=H(rk，rm)。故障檢測與恢復的偽代碼如算法2 所示。

算法2故障檢測與恢復算法

在隨機選取子集群中的兩個reduce 節點后，每次TMR 算法驗證都需發送16 份中間結果給reduce 節點，假設每個中間結果大小為τ，那么每個CTMR 模型在驗證過程中需要發送16τ的數據。在含有N個節點的分布式集群中，共需發送的數據量。two-stage TMR 算法在最優情況下只需發送8τ的數據量，即最初選擇的兩個副本對應的中間結果一致，在最壞情況下所有數據最初選擇的兩個副本的對應中間結果均不一致，因此需要第3 個副本進行多數一致表決，共需發送16τ的數據量。在CTMR 算法中，每個節點使用分布式預編碼本地計算的中間結果減少通信開銷。在最優情況下，每個模型只需8τ的通信量，即模型最初選擇兩個校驗點即可得到正確結果。在最壞情況下，每個模型需要16τ的通信量，即在最初選擇兩個校驗節點后不能得到正確結果，需更換節點重新做校驗。因此，在包含N個節點的分布式集群中，共需發送的數據量。CTMR算法與TMR算法的通信開銷之比如式（8）所示，即CTMR算法總能在小于等于TMR 算法通信量的情況下得到正確結果。

2.3 算法實例

如圖2 所示，在含有N=6 個節點的分布式集群中，輸入系統數據量為M=8，將該數據集劃分為8 個數據塊{b1，b2，…，b8}。每個節點在map 階段計算本地數據集產生相應的中間結果集，例如node2在map 階段針對本地數據集B2={b1，b2，b5，b6}計算得到相應的中間結果集。考慮node1在map 階段由于瞬態故障導致計算錯誤情況下的容錯過程，首先選取node1和node6作為校驗節點，子集群中其余4 個節點為node2、node3、node4、node5，將4 個節點與node1冗余存儲的數據集R2，1、R3，1、R4，1、R5，1所對應的中間結果分別使用分布式編碼計算得到相應的編碼結果u2，1、u3，1、u4，1、u5，1發送給node1，將4 個節點與node6冗余存儲的數據集R2，6、R3，6、R4，6、R5，6所對應的中間結果分別使用分布式編碼計算得到相應的編碼結果u2，6、u3，6、u4，6、u5，6發送給node6，其中，。

圖2 node1中 和 錯誤時的容錯過程Fig.2 Fault-tolerant process of and incorrect in the node1

3 實驗結果與分析

3.1 實驗方案

本文分布式計算的測試程序為Terasort［22］，CTMR算法的評價指標為任務執行總時間、map 和shuffle 階段執行時間以及平均故障修復時間（Mean Time to Repair，MTTR），對比算法為TMR 和two-stage TMR算法。

實驗使用多臺虛擬機搭建的分布式集群，包括1個管理節點和6 個工作節點，節點間的帶寬為100 Mb/s。實驗中動態選擇發生故障的節點個數，隨機選取節點并修改其對應數據塊數值實現故障注入。假設系統在單位時間內的故障發生概率服從泊松分布p(x=k)=，即在單位時間內出現k個故障的概率為p(k)，本文中λ的取值為2。因此，在滿足泊松分布的條件下，假設該分布式系統隨機產生k個故障。如果各個故障之間相互獨立，那么在連續運行的分布式系統中，當產生k個故障時的無故障運行時間為tk，則系統平均無故障運行時間如式（9）所示：

故障修復時間為檢測到故障直至故障修復的時間，假設有k個故障時的故障修復時間為θk，則平均故障修復時間如式（10）所示：

實驗中每個MapReduce任務可以分為map、shuffle、check 和reduce 這4 個階段。在map 階段，管理節點按照CTMR 算法要求將用戶輸入數據分發給6 個工作節點，同時指定2 個校驗節點。每個工作節點對本地數據集進行排序，得到相應的中間結果集。在shuffle 階段，每個節點將其中間結果編碼，發送給之前指定的校驗節點。在check 階段，校驗節點將收到的數據包進行故障檢測和恢復，校驗成功后得到相應結果。在reduce階段，管理節點收到校驗節點發送來的部分reduce 計算結果后執行reduce 函數得到最終輸出結果。任務執行總時間Ttotal如式（11）所示：

3.2 實驗結果

圖3 給出了CTMR、two-stage TMR 以及TMR 算法的任務執行總時間對比結果?？梢钥闯觯珻TMR算法能有效降低分布式計算的任務執行總時間。當故障個數較少時，CTMR 算法的執行效率遠高于另外兩種算法。隨著故障個數的不斷增加，two-stage TMR 算法由于需要重新執行第3 個副本做驗證，而CTMR 算法則需要更換節點做驗證，在該過程中需要重新發送編碼數據包，因此這兩種算法的任務執行總時間也會隨之增加。

圖3 任務執行總時間對比Fig.3 Comparison of total task execution time

圖4 給出了CTMR、two-stage TMR 以及TMR 算法在map 階段的執行時間對比結果。可以看出，隨著故障個數的增加，two-stage TMR 算法由于第1 次投票選擇的2 個副本對應的中間結果不同，因此需要進行第2 次投票。這時會選擇第3 個副本并執行map 任務，因此map 階段所需時間隨故障個數的增加不斷增加。TMR 和CTMR 算法由于最初都要對3 個副本執行map 任務，因此map 階段的執行時間基本保持不變。

圖4 map 階段執行時間對比Fig.4 Comparison of the execution time in the map phase

圖5 給出了CTMR、two-stage TMR 以及TMR 算法在shuffle 階段的執行時間對比結果?？梢钥闯?，CTMR 算法在shuffle 階段所需時間明顯低于TMR算法，并且相比two-stage TMR 算法有一定程度的減少。本文將shuffle 階段的執行時間作為通信開銷的衡量指標，當系統在單位時間內的故障發生概率服從泊松分布時，可以計算得出發生k個故障的概率為p（k），本文中λ的取值為2。根據式（9）可計算出CTMR、two-stage TMR 以及TMR 算法在shuffle 階段的執行時間分別為1.90 s、2.21 s 和3.22 s。因此，CTMR 算法在shuffle 階段的執行時間相比TMR 算法降低了41.0%，相比two-stage TMR 算法降低了14.0%。

圖6 給出了CTMR、two-stage TMR 以及TMR 算法在check 階段的執行時間對比結果?？梢钥闯觯珻TMR 算法在一定的故障個數范圍內，故障修復效率明顯優于TMR 與two-stage TMR 算法。隨著故障個數的不斷增加，TMR 和two-stage TMR 算法均需要對所有副本進行第3 次投票，而CTMR 算法也需要更換節點進行校驗，因此3 種算法的故障恢復時間不斷增加并最終趨于一致。

圖6 check 階段執行時間對比Fig.6 Comparison of the execution time in the check phase

當系統在單位時間內發生故障的概率服從泊松分布時，根據式（9）分別計算出CTMR、two-stage TMR 以及TMR 算法在tatal 階段的任務執行總時間以及map、shuffle 和check 階段的任務執行時間，結果如圖7 所示?？梢钥闯?，CTMR 算法的任務執行總時間相比TMR 算法降低了25.8%，相比two-stage TMR 算法降低了13.2%。根據式（10），CTMR 算法的平均故障修復時間相比TMR 算法降低了18.3%，相比two-stage TMR 算法降低了26.2%。

圖7 故障發生概率服從泊松分布時3 種算法的執行時間對比Fig.7 Comparison of the execution time of the three algorithms when the probability of failure obeys the Poisson distribution

4 結束語

為降低MapReduce 分布式計算中容錯算法的通信開銷，本文結合副本冗余技術和分布式編碼計算技術，提出一種新的容錯算法。實驗結果表明，CTMR 算法在完成容錯計算的同時，相比TMR 和two-stage TMR 容錯算法，平均降低了41.0% 和14.0%的shuffle 階段的通信開銷以及18.3%和26.2%的平均故障修復時間，并且提高了分布式系統的可用性和可靠性。但由于本文中的副本數量固定為3具有一定的局限性，因此下一步將根據分布式系統的故障發生概率，通過動態調整副本數量以增強容錯算法的靈活性。