面向磁盤(pán)駐留的類(lèi)Pregel系統(tǒng)的多級(jí)容錯(cuò)處理機(jī)制

畢亞輝姜蘇洋王志剛冷芳玲鮑玉斌于 戈錢(qián) 嶺

1(東北大學(xué)計(jì)算機(jī)科學(xué)與工程學(xué)院 沈陽(yáng) 110819)2(中國(guó)移動(dòng)(蘇州)軟件技術(shù)有限公司 江蘇蘇州 215163)(biyahui1990@163.com)

?

面向磁盤(pán)駐留的類(lèi)Pregel系統(tǒng)的多級(jí)容錯(cuò)處理機(jī)制

畢亞輝1姜蘇洋1王志剛1冷芳玲1鮑玉斌1于 戈1錢(qián) 嶺2

1(東北大學(xué)計(jì)算機(jī)科學(xué)與工程學(xué)院 沈陽(yáng) 110819)2(中國(guó)移動(dòng)(蘇州)軟件技術(shù)有限公司 江蘇蘇州 215163)(biyahui1990@163.com)

基于BSP模型的分布式框架已經(jīng)成為大規(guī)模圖高頻迭代處理的有效工具.分布式系統(tǒng)可以通過(guò)增加集群節(jié)點(diǎn)數(shù)量的方式提供彈性的處理能力，但同時(shí)也增加了故障發(fā)生的概率，因此亟需開(kāi)發(fā)高效的容錯(cuò)處理機(jī)制.現(xiàn)有工作主要是基于檢查點(diǎn)機(jī)制展開(kāi)研究，包括數(shù)據(jù)備份和故障恢復(fù)2部分：前者沒(méi)有考慮迭代過(guò)程中參與計(jì)算的數(shù)據(jù)規(guī)模的動(dòng)態(tài)變化，而是備份所有圖數(shù)據(jù)，因此引入了冗余數(shù)據(jù)的寫(xiě)開(kāi)銷(xiāo)；后者通常是從遠(yuǎn)程存儲(chǔ)節(jié)點(diǎn)上讀取備份數(shù)據(jù)進(jìn)行故障恢復(fù)，而沒(méi)有考慮利用本地磁盤(pán)數(shù)據(jù)恢復(fù)某些場(chǎng)景下的故障，引入額外的網(wǎng)絡(luò)開(kāi)銷(xiāo).因此提出了一種多級(jí)容錯(cuò)處理機(jī)制，將故障分為計(jì)算任務(wù)故障和計(jì)算節(jié)點(diǎn)故障2類(lèi)，并設(shè)計(jì)了不同的備份和恢復(fù)策略. 備份階段利用了某些應(yīng)用在迭代計(jì)算過(guò)程中參與計(jì)算的數(shù)據(jù)規(guī)模的動(dòng)態(tài)變化特性，設(shè)計(jì)了完全備份和寫(xiě)變化log自適應(yīng)選擇的策略，可以顯著減少冗余數(shù)據(jù)的寫(xiě)開(kāi)銷(xiāo).故障恢復(fù)階段，對(duì)任務(wù)故障，利用本地磁盤(pán)上保留的圖數(shù)據(jù)和遠(yuǎn)程的消息數(shù)據(jù)完成恢復(fù)；而對(duì)節(jié)點(diǎn)故障，則利用備份在遠(yuǎn)程信息進(jìn)行恢復(fù).最后，通過(guò)在真實(shí)數(shù)據(jù)集上的大量實(shí)驗(yàn),驗(yàn)證了提出的多級(jí)容錯(cuò)機(jī)制的有效性.

容錯(cuò)；大規(guī)模圖；迭代計(jì)算；BSP模型；檢查點(diǎn)

隨著圖數(shù)據(jù)規(guī)模的快速增長(zhǎng)和分析復(fù)雜性的不斷增加，大量支持大規(guī)模圖迭代計(jì)算的分布式處理系統(tǒng)被開(kāi)發(fā)[1-3]，其中，Giraph[1]和BC-BSP[2]在迭代計(jì)算過(guò)程中提供了基于磁盤(pán)輔助的數(shù)據(jù)和中間消息存儲(chǔ).分布式系統(tǒng)可以通過(guò)增加計(jì)算節(jié)點(diǎn)數(shù)量的方式提高處理的能力和效率.然而，系統(tǒng)在迭代過(guò)程中發(fā)生故障的概率與節(jié)點(diǎn)規(guī)模成正比[4].對(duì)于長(zhǎng)時(shí)間迭代計(jì)算的圖處理應(yīng)用，需要設(shè)計(jì)高效的容錯(cuò)處理機(jī)制.

目前分布式圖處理系統(tǒng)采取的處理故障方法一般是基于檢查點(diǎn)的方法[2-3].檢查點(diǎn)機(jī)制包括數(shù)據(jù)備份與數(shù)據(jù)恢復(fù)2部分.各個(gè)任務(wù)周期性地將圖數(shù)據(jù)和有關(guān)信息備份到分布式文件系統(tǒng)(如HDFS)中.當(dāng)系統(tǒng)發(fā)生故障時(shí)，各任務(wù)從分布式文件系統(tǒng)中讀取檢查點(diǎn)備份的數(shù)據(jù)和有關(guān)信息來(lái)完成故障恢復(fù).基于檢查點(diǎn)的方法原理簡(jiǎn)單明了，容易實(shí)現(xiàn).然而，現(xiàn)有的基于檢查點(diǎn)的方法存在2方面的不足：

1) 在數(shù)據(jù)備份時(shí)，并沒(méi)有區(qū)分迭代過(guò)程中數(shù)據(jù)是否發(fā)生動(dòng)態(tài)變化，而是將所有的圖數(shù)據(jù)信息進(jìn)行備份，因此導(dǎo)致寫(xiě)檢查點(diǎn)時(shí)產(chǎn)生了大量冗余數(shù)據(jù)的寫(xiě)操作；

2) 在故障恢復(fù)階段，通常是從遠(yuǎn)程存儲(chǔ)節(jié)點(diǎn)上讀取備份的信息進(jìn)行故障恢復(fù)，而沒(méi)有考慮利用本地磁盤(pán)數(shù)據(jù)恢復(fù)某些場(chǎng)景下的故障，尤其是在面向磁盤(pán)駐留的計(jì)算系統(tǒng)中，例如任務(wù)故障的情形，使得在某些類(lèi)型的故障恢復(fù)過(guò)程中需要遠(yuǎn)程讀取檢查點(diǎn)數(shù)據(jù)，存在“遠(yuǎn)程讀”問(wèn)題，引入了網(wǎng)絡(luò)開(kāi)銷(xiāo).

針對(duì)上述2個(gè)問(wèn)題，本文提出了一種面向磁盤(pán)駐留的類(lèi)Pregel系統(tǒng)的多級(jí)容錯(cuò)處理機(jī)制.所謂的多級(jí)是針對(duì)任務(wù)故障和節(jié)點(diǎn)故障的數(shù)據(jù)備份與恢復(fù)策略而言的.首先，對(duì)于數(shù)據(jù)備份策略，根據(jù)數(shù)據(jù)備份的位置和規(guī)模，可以分為3個(gè)級(jí)別:第1級(jí)別，被處理的圖數(shù)據(jù)有本地備份和消息數(shù)據(jù)在HDFS上備份；第2級(jí)別，靜態(tài)數(shù)據(jù)(頂點(diǎn)的出度鄰接表，并假設(shè)在迭代計(jì)算過(guò)程中不改變)、動(dòng)態(tài)數(shù)據(jù)(迭代過(guò)程中動(dòng)態(tài)變化，如PageRank的PR值)和消息都備份在HDFS上；第3級(jí)別，HDFS上備份有圖的動(dòng)態(tài)數(shù)據(jù)加日志(log)信息、靜態(tài)數(shù)據(jù)和消息.對(duì)應(yīng)于數(shù)據(jù)備份的3個(gè)級(jí)別，故障的恢復(fù)也有3個(gè)級(jí)別：第1級(jí)別，讀取本地的數(shù)據(jù)和HDFS上的消息；第2級(jí)別，讀取HDFS上的靜態(tài)數(shù)據(jù)、動(dòng)態(tài)數(shù)據(jù)和消息；第3級(jí)別，讀取HDFS上的靜態(tài)數(shù)據(jù)、啟用log機(jī)制(記錄變化的動(dòng)態(tài)數(shù)據(jù))后的動(dòng)態(tài)數(shù)據(jù)和消息.對(duì)于任務(wù)故障的備份與恢復(fù)采用的是第1級(jí)別的容錯(cuò)處理機(jī)制，對(duì)于節(jié)點(diǎn)故障備份與恢復(fù)采用的是第2級(jí)別和第3級(jí)別的容錯(cuò)處理機(jī)制.本文所提出的多級(jí)容錯(cuò)處理機(jī)制能夠有效地處理分布式圖處理系統(tǒng)出現(xiàn)的任務(wù)故障以及節(jié)點(diǎn)故障，該機(jī)制適用于采用磁盤(pán)輔助的類(lèi)Pregel系統(tǒng).

本文的主要貢獻(xiàn)如下：

1) 任務(wù)故障的恢復(fù)直接讀取本地磁盤(pán)的靜態(tài)數(shù)據(jù)與動(dòng)態(tài)數(shù)據(jù)和HDFS上的消息，避免了加載HDFS上的靜態(tài)和動(dòng)態(tài)數(shù)據(jù)的開(kāi)銷(xiāo)，加快了任務(wù)故障的處理過(guò)程.

2) 提出了log機(jī)制，當(dāng)參與計(jì)算的圖數(shù)據(jù)規(guī)模小于指定閾值時(shí)，使用log方式記錄數(shù)據(jù)變化而不是全部備份動(dòng)態(tài)數(shù)據(jù)，以減少備份數(shù)據(jù)的寫(xiě)開(kāi)銷(xiāo).此外，本文還給出了log啟動(dòng)閾值設(shè)置的理論分析.

3) 在大量的實(shí)驗(yàn)基礎(chǔ)上，對(duì)比了傳統(tǒng)的檢查點(diǎn)(checkpoint)機(jī)制和本文的多級(jí)容錯(cuò)處理機(jī)制，驗(yàn)證了本文的多級(jí)容錯(cuò)處理機(jī)制的有效性.

1 相關(guān)工作

設(shè)計(jì)高效的容錯(cuò)方案始終是分布式圖處理系統(tǒng)重點(diǎn)解決的問(wèn)題.因此，已有許多關(guān)于分布式(圖)處理系統(tǒng)的容錯(cuò)機(jī)制的研究工作.已有的方法可以分為基于檢查點(diǎn)的方法、基于日志的方法和混合方法3類(lèi).

目前大多數(shù)知名的分布式圖處理系統(tǒng)如Giraph[1], GraphLab[5], PowerGraph[6], GPS[7], Mizan[8]系統(tǒng)采用的都是基于傳統(tǒng)檢查點(diǎn)的方法;GraphX[9]采用基于日志的方法.Pregel[3]系統(tǒng)中提供了2種容錯(cuò)機(jī)制：1)基本的寫(xiě)檢查點(diǎn)機(jī)制;2)受限的恢復(fù)機(jī)制，即采用的是一種基于檢查點(diǎn)和日志相結(jié)合的混合方法.

Pregel的基于基本寫(xiě)檢查點(diǎn)機(jī)制實(shí)現(xiàn)的容錯(cuò)機(jī)制是周期性地備份頂點(diǎn)的狀態(tài)和消息以實(shí)現(xiàn)容錯(cuò).當(dāng)一個(gè)或多個(gè)節(jié)點(diǎn)發(fā)生故障，主節(jié)點(diǎn)重新分配這些圖的分區(qū)到當(dāng)前可用的工作節(jié)點(diǎn)集合上，這些節(jié)點(diǎn)會(huì)從最近記錄檢查點(diǎn)的超步S開(kāi)始重新加載分區(qū)狀態(tài).

Pregel提出的另一種受限的容錯(cuò)恢復(fù)機(jī)制是一種基于檢查點(diǎn)和基于日志相結(jié)合的方法.除了基本的檢查點(diǎn)，工作節(jié)點(diǎn)同時(shí)將圖數(shù)據(jù)加載和迭代計(jì)算期間從這個(gè)節(jié)點(diǎn)上分區(qū)發(fā)出去的消息記錄到日志中，這樣故障恢復(fù)就會(huì)被限制在丟失的分區(qū)上.這種方法的優(yōu)點(diǎn)是：只重新計(jì)算丟失的分區(qū)，節(jié)省了恢復(fù)時(shí)的計(jì)算資源，同時(shí)由于每個(gè)工作節(jié)點(diǎn)需要恢復(fù)的分區(qū)很少，減少了恢復(fù)的延遲；缺點(diǎn)是對(duì)發(fā)送出去的消息進(jìn)行保存會(huì)產(chǎn)生一定的存儲(chǔ)開(kāi)銷(xiāo)，降低了作業(yè)正常運(yùn)行時(shí)的效率.本文的恢復(fù)機(jī)制雖然還是要重新計(jì)算所有分區(qū)，但通過(guò)日志記錄發(fā)生變化的動(dòng)態(tài)數(shù)據(jù)可以減少檢查點(diǎn)的存儲(chǔ)開(kāi)銷(xiāo)及網(wǎng)絡(luò)IO開(kāi)銷(xiāo).Pregel的受限恢復(fù)機(jī)制可以與本文的工作互補(bǔ).

Spark系統(tǒng)[10]將圖數(shù)據(jù)信息分為動(dòng)態(tài)數(shù)據(jù)和靜態(tài)數(shù)據(jù).寫(xiě)檢查點(diǎn)只記錄動(dòng)態(tài)變化的部分.對(duì)于絕大部分真實(shí)圖，靜態(tài)數(shù)據(jù)的規(guī)模遠(yuǎn)大于動(dòng)態(tài)數(shù)據(jù)，因此這種方式極大減少了寫(xiě)檢查點(diǎn)的開(kāi)銷(xiāo).本文的多級(jí)容錯(cuò)機(jī)制借鑒了Spark的這種處理方式，即第2級(jí)別.進(jìn)一步地，對(duì)于某些算法，如單源最短路徑(SSSP)，迭代過(guò)程中僅有部分頂點(diǎn)參與計(jì)算，即參與更新計(jì)算的動(dòng)態(tài)數(shù)據(jù)的規(guī)模是變化的.針對(duì)這種情形，本文提出了第3級(jí)容錯(cuò)方案——寫(xiě)日志機(jī)制(即log機(jī)制)來(lái)進(jìn)一步減少I(mǎi)O開(kāi)銷(xiāo).此外，Spark對(duì)于任務(wù)故障和節(jié)點(diǎn)故障的恢復(fù)都是加載存儲(chǔ)在分布式文件系統(tǒng)的檢查點(diǎn)數(shù)據(jù)，沒(méi)有利用本地磁盤(pán)數(shù)據(jù).

GraphX[4]采用的是基于日志(血統(tǒng))的恢復(fù)方法，它利用彈性分布式數(shù)據(jù)集(RDD)加速故障恢復(fù).然而，當(dāng)一個(gè)節(jié)點(diǎn)發(fā)生故障時(shí)，這個(gè)節(jié)點(diǎn)上的圖數(shù)據(jù)仍然需要恢復(fù).

文獻(xiàn)[11]則針對(duì)傳統(tǒng)檢查點(diǎn)性能低下的問(wèn)題提出了基于內(nèi)存緩存的異步檢查點(diǎn)容錯(cuò)方法.其主要思想是將檢查點(diǎn)臨時(shí)緩存在節(jié)點(diǎn)的內(nèi)存中，然后由另一個(gè)輔助任務(wù)將緩存在內(nèi)存中的檢查點(diǎn)數(shù)據(jù)寫(xiě)到分布式文件系統(tǒng).但是這種異步的檢查點(diǎn)容錯(cuò)方法并不適用于類(lèi)Pregel系統(tǒng)，因?yàn)轭?lèi)Pregel系統(tǒng)需要在寫(xiě)檢查點(diǎn)時(shí)進(jìn)行全局同步才能進(jìn)入下一個(gè)超步.

2 多級(jí)容錯(cuò)處理機(jī)制概述

本節(jié)首先介紹BC-BSP系統(tǒng)及其現(xiàn)有的檢查點(diǎn)機(jī)制，然后介紹本文的備份與恢復(fù)框架.

2.1 BC-BSP系統(tǒng)簡(jiǎn)介

BC-BSP系統(tǒng)[2]是基于BSP模型的開(kāi)源大圖迭代處理系統(tǒng)，支持多種數(shù)據(jù)輸入方式和使用磁盤(pán)輔助暫存數(shù)據(jù)(簡(jiǎn)稱(chēng)磁盤(pán)操作)，具有良好的容錯(cuò)控制能力和可伸縮性.圖1給出了BC-BSP的系統(tǒng)結(jié)構(gòu)圖.它包括客戶(hù)端(Client)、BSP Controller端、Worker端、Staff端和完成同步協(xié)調(diào)的ZooKeeper.

客戶(hù)端是用戶(hù)與BC-BSP系統(tǒng)交互的實(shí)體，作業(yè)的提交和運(yùn)行狀態(tài)的監(jiān)控均需要通過(guò)客戶(hù)端平臺(tái)實(shí)現(xiàn).Controller端是BC-BSP系統(tǒng)的中樞控制系統(tǒng)，負(fù)責(zé)調(diào)控整個(gè)集群，包括作業(yè)調(diào)度、故障恢復(fù)等.Worker端是工作節(jié)點(diǎn)的控制中心，隸屬于Controller端，負(fù)責(zé)本節(jié)點(diǎn)的整體運(yùn)行調(diào)控.Staff端是工作實(shí)體，完成具體的工作任務(wù)，從邏輯上講，按照用戶(hù)提交的作業(yè)進(jìn)行組織，但是在集群中受Worker端的直接管理.全局同步、消息通信和容錯(cuò)控制，是作業(yè)運(yùn)行過(guò)程中的重要環(huán)節(jié)，需要Controller端、Worker端和Staff端的協(xié)同工作來(lái)實(shí)現(xiàn).其中的ZooKeeper作為第三方插件，在BC-BSP系統(tǒng)的任務(wù)調(diào)度模塊、高可用(HA)管理模塊、全局同步模塊以及聚集計(jì)算功能的實(shí)現(xiàn)中具有重要作用.

Fig. 1 The system structure of BC-BSP.圖1 BC-BSP系統(tǒng)結(jié)構(gòu)關(guān)系圖

鑒于數(shù)據(jù)量的不斷激增和硬件資源的相對(duì)缺乏，BC-BSP系統(tǒng)支持使用磁盤(pán)作為迭代計(jì)算過(guò)程中的輔助存儲(chǔ)介質(zhì)，暫存圖數(shù)據(jù)和中間消息數(shù)據(jù)等，而不是假設(shè)所有數(shù)據(jù)(包括中間的消息數(shù)據(jù))都在內(nèi)存.因此，系統(tǒng)中實(shí)現(xiàn)了磁盤(pán)緩存模塊，它負(fù)責(zé)暫存系統(tǒng)計(jì)算時(shí)內(nèi)存無(wú)法容納的圖數(shù)據(jù)和消息數(shù)據(jù).其基本思路是：對(duì)于圖數(shù)據(jù)，在迭代計(jì)算過(guò)程中常駐磁盤(pán)，在圖處理系統(tǒng)的數(shù)據(jù)加載階段，每個(gè)計(jì)算任務(wù)從原始數(shù)據(jù)所在的存儲(chǔ)系統(tǒng)(通常為HDFS或HBase)按照數(shù)據(jù)分片記錄的位置信息加載數(shù)據(jù)，數(shù)據(jù)加載程序每讀取一個(gè)頂點(diǎn)的數(shù)據(jù)，就按照該頂點(diǎn)的ID值，根據(jù)系統(tǒng)設(shè)定的映射規(guī)則，將其寫(xiě)入到對(duì)應(yīng)節(jié)點(diǎn)的磁盤(pán)塊中.圖數(shù)據(jù)在本地磁盤(pán)的存儲(chǔ)是按照Hash分桶組織，且每個(gè)任務(wù)的數(shù)據(jù)被分成圖頂點(diǎn)(動(dòng)態(tài)數(shù)據(jù))、邊(靜態(tài)數(shù)據(jù))和消息3個(gè)部分，每一部分都分為若干個(gè)Hash桶存放到本地磁盤(pán)上，桶的數(shù)量可由用戶(hù)自行設(shè)定.進(jìn)入迭代計(jì)算階段，每個(gè)超步結(jié)束后，將動(dòng)態(tài)變化的頂點(diǎn)數(shù)據(jù)寫(xiě)回本地磁盤(pán)，而不發(fā)生變化的靜態(tài)數(shù)據(jù)只在需要處理時(shí)才從本地磁盤(pán)讀入內(nèi)存，處理結(jié)束后并不需要寫(xiě)回磁盤(pán)，因?yàn)樗鼪](méi)有變化.而對(duì)消息數(shù)據(jù)，則盡可能地存儲(chǔ)在內(nèi)存中，如消息發(fā)送時(shí)內(nèi)存緩沖區(qū)中的數(shù)據(jù)量超出用戶(hù)設(shè)置的緩沖區(qū)上限，計(jì)算等待發(fā)送；在消息接收時(shí)，如果所占用緩沖區(qū)的大小也超出用戶(hù)設(shè)置的接收消息緩沖區(qū)上限，則接收過(guò)程要同步等待數(shù)據(jù)塊寫(xiě)入磁盤(pán).

2.2 BC-BSP現(xiàn)有的容錯(cuò)機(jī)制

BC-BSP當(dāng)前版本的數(shù)據(jù)備份就是對(duì)作業(yè)本地計(jì)算的中間結(jié)果按照一定的頻率(比如每隔k個(gè)超步)記錄檢查點(diǎn).分布式文件系統(tǒng)中記錄的檢查點(diǎn)由3部分信息組成：1)原始的圖數(shù)據(jù)信息，該部分?jǐn)?shù)據(jù)在作業(yè)完成之前一直存在；2)每次以增量方式(即只記錄頂點(diǎn)動(dòng)態(tài)數(shù)據(jù)而不記錄頂點(diǎn)的出邊信息)記錄的檢查點(diǎn)信息；3)各個(gè)分區(qū)收到的、在下個(gè)超步處理的消息.為了節(jié)省存儲(chǔ)資源，當(dāng)新的檢查點(diǎn)記錄成功之后則刪除歷史檢查點(diǎn).數(shù)據(jù)恢復(fù)即從分布式文件系統(tǒng)加載最后記錄的檢查點(diǎn)信息，加載時(shí)要同時(shí)讀取原始圖數(shù)據(jù)信息和最近的增量檢查點(diǎn)信息，以及備份的消息這樣才能還原到最近檢查點(diǎn)記錄時(shí)圖處理作業(yè)繼續(xù)運(yùn)行的上下文狀態(tài).BC-BSP系統(tǒng)寫(xiě)檢查點(diǎn)的流程如圖2所示.我們稱(chēng)這種容錯(cuò)機(jī)制為“增量檢查點(diǎn)”機(jī)制.

Fig. 2 The flowchart of write checkpoint.圖2 寫(xiě)檢查點(diǎn)流程圖

BC-BSP系統(tǒng)對(duì)故障的檢測(cè)是通過(guò)心跳機(jī)制完成的.當(dāng)主節(jié)點(diǎn)在一定的時(shí)間內(nèi)沒(méi)有收到工作節(jié)點(diǎn)的心跳信息，就把該節(jié)點(diǎn)標(biāo)記為故障節(jié)點(diǎn).

2.3 多級(jí)容錯(cuò)機(jī)制的備份與恢復(fù)框架

系統(tǒng)運(yùn)行過(guò)程中各個(gè)任務(wù)加載分區(qū)數(shù)據(jù)到該任務(wù)本地的磁盤(pán)上，動(dòng)態(tài)數(shù)據(jù)每個(gè)迭代步都寫(xiě)回本地磁盤(pán)，靜態(tài)數(shù)據(jù)在每次迭代計(jì)算中是只讀的.進(jìn)入迭代計(jì)算階段，如果沒(méi)有發(fā)生故障，無(wú)論動(dòng)態(tài)數(shù)據(jù)或是靜態(tài)數(shù)據(jù)的訪(fǎng)問(wèn)都是針對(duì)本地磁盤(pán)的.迭代過(guò)程中，系統(tǒng)按照配置文件中設(shè)置的檢查點(diǎn)頻率周期性地記錄檢查點(diǎn).在增量檢查點(diǎn)機(jī)制中，除了第1次寫(xiě)檢查點(diǎn)時(shí)需要記錄完整的圖數(shù)據(jù)(包括頂點(diǎn)Id、value值和出邊)之外，其后的每個(gè)檢查點(diǎn)只需記錄圖的動(dòng)態(tài)數(shù)據(jù)(頂點(diǎn)Id與value值)即可.若計(jì)算過(guò)程中存在節(jié)點(diǎn)間交互，則這種交互的信息都以消息的形式備份到HDFS上.故障恢復(fù)時(shí)會(huì)讀取檢查點(diǎn)及備份的消息進(jìn)行恢復(fù).但是，通過(guò)對(duì)某些應(yīng)用的運(yùn)行特征進(jìn)行觀察，我們發(fā)現(xiàn)圖的動(dòng)態(tài)部分也不是全部變化的，例如在單源最短路徑計(jì)算中每次參與計(jì)算的點(diǎn)很少.因此，當(dāng)動(dòng)態(tài)數(shù)據(jù)變化的規(guī)模小于一定閾值時(shí)，啟用log機(jī)制來(lái)記錄變化的動(dòng)態(tài)數(shù)據(jù)，這樣需要備份的數(shù)據(jù)量就小于完整的動(dòng)態(tài)數(shù)據(jù)部分.這里的關(guān)鍵是閾值的確定問(wèn)題，3.1節(jié)將詳細(xì)討論.多級(jí)容錯(cuò)機(jī)制備份算法如算法1所示.

算法1.computeFramework().

輸入：log機(jī)制啟用標(biāo)志logFlag.

① Whileflag=true /*flag：本地循環(huán)計(jì)算標(biāo)志*/

② For each vertexv

③compute()；

④ IflogFlag=true

⑤ 將v放到c中；/*c：值發(fā)生變化的頂集合*/

⑥ End If

⑦ End For

⑧ 將動(dòng)態(tài)數(shù)據(jù)寫(xiě)回本地磁盤(pán)文件；

⑨ IfcommandType.equals(“CHECKPOINT”)

/*commandType：超步命令類(lèi)型*/

⑩ 將消息寫(xiě)到HDFS；

當(dāng)故障發(fā)生時(shí)，針對(duì)不同的故障類(lèi)型采取不同的恢復(fù)策略.故障恢復(fù)過(guò)程的框架見(jiàn)算法2所示.

算法2.FaultRecovery(faultType).

輸入：故障類(lèi)型faultType.

① IffaultType.equals(“任務(wù)故障”)

② 加載本地靜態(tài)和動(dòng)態(tài)數(shù)據(jù)及HDFS上的消息；

③ElseIflogFlag=true/*logFlag：log啟用的標(biāo)志*/

④ 從HDFS加載檢查點(diǎn)數(shù)據(jù)、日志和消息；

⑤ Else

⑥ 從HDFS加載檢查點(diǎn)數(shù)據(jù)和消息；

⑦ End If

對(duì)于任務(wù)故障，系統(tǒng)直接在本地重啟故障的任務(wù)，各個(gè)任務(wù)(包括重啟的恢復(fù)任務(wù))直接利用本地保存的圖數(shù)據(jù)以及遠(yuǎn)程的消息數(shù)據(jù)恢復(fù)到最近的檢查點(diǎn)，因?yàn)閳D數(shù)據(jù)在本地有完整的信息，且任務(wù)故障不會(huì)造成本地的數(shù)據(jù)不可用(除了極少數(shù)文件損壞的情況外).這就避免了加載HDFS上的檢查點(diǎn)圖數(shù)據(jù)，從一定程度上加快了任務(wù)恢復(fù)的過(guò)程.而對(duì)于節(jié)點(diǎn)故障，系統(tǒng)首先利用故障恢復(fù)調(diào)度機(jī)制對(duì)在這個(gè)節(jié)點(diǎn)上的所有任務(wù)進(jìn)行遷移操作，因?yàn)楣?jié)點(diǎn)發(fā)生故障就不能再使用存儲(chǔ)在本地的圖數(shù)據(jù)進(jìn)行恢復(fù)了.此時(shí)，如果發(fā)生故障的任務(wù)沒(méi)有啟用log機(jī)制，那么遷移后的任務(wù)通過(guò)讀取HDFS上的靜態(tài)數(shù)據(jù)、動(dòng)態(tài)數(shù)據(jù)和消息進(jìn)行恢復(fù)；如故障任務(wù)啟用了log機(jī)制，通過(guò)讀取HDFS上的靜態(tài)數(shù)據(jù)、log機(jī)制記錄的動(dòng)態(tài)數(shù)據(jù)和消息進(jìn)行恢復(fù).

3 多級(jí)容錯(cuò)機(jī)制的數(shù)據(jù)備份策略

寫(xiě)檢查點(diǎn)是常用的容錯(cuò)數(shù)據(jù)備份機(jī)制：按照一定的頻率或超步間隔將各個(gè)任務(wù)處理的數(shù)據(jù)和頂點(diǎn)所收到的消息寫(xiě)入分布式存儲(chǔ)介質(zhì)(如HDFS).因?yàn)榧僭O(shè)內(nèi)存不足，系統(tǒng)所處理的數(shù)據(jù)常駐磁盤(pán)，需要時(shí)才加載到內(nèi)存，所以各任務(wù)處理的數(shù)據(jù)每個(gè)超步結(jié)束后都保存到本地磁盤(pán)，靜態(tài)部分常駐磁盤(pán)，動(dòng)態(tài)變化部分每個(gè)超步都寫(xiě)回本地磁盤(pán).

這樣，利用本地的靜態(tài)數(shù)據(jù)、動(dòng)態(tài)數(shù)據(jù)和HDFS上備份的消息就可以完成任務(wù)故障的恢復(fù).

3.1 log機(jī)制及其啟用條件

在現(xiàn)有的增量備份策略中，是將動(dòng)態(tài)部分?jǐn)?shù)據(jù)全備份到遠(yuǎn)程，但實(shí)際上有些應(yīng)用每次迭代計(jì)算，甚至在寫(xiě)檢查點(diǎn)間隔期間，并不會(huì)更新分區(qū)上所有的狀態(tài)或者值，因此為了減少寫(xiě)入檢查點(diǎn)的冗余數(shù)據(jù)，當(dāng)頂點(diǎn)值發(fā)生變化的比例低于一定的閾值時(shí)，就開(kāi)啟log機(jī)制.所謂的log機(jī)制就是在迭代過(guò)程中只有一小部分頂點(diǎn)的值發(fā)生改變時(shí)，記錄這些發(fā)生改變的頂點(diǎn)的信息.log機(jī)制的實(shí)現(xiàn)可以有2種方式：作業(yè)級(jí)的實(shí)現(xiàn)和任務(wù)級(jí)的實(shí)現(xiàn).作業(yè)級(jí)的實(shí)現(xiàn)就是當(dāng)作業(yè)滿(mǎn)足開(kāi)啟log機(jī)制的條件時(shí)，對(duì)這個(gè)作業(yè)的所有任務(wù)都啟用log機(jī)制；任務(wù)級(jí)實(shí)現(xiàn)是針對(duì)某個(gè)任務(wù)而言的，如某個(gè)任務(wù)滿(mǎn)足啟用log機(jī)制的條件，對(duì)這個(gè)任務(wù)本身啟用log機(jī)制.log機(jī)制的作業(yè)級(jí)實(shí)現(xiàn)的優(yōu)點(diǎn)是：當(dāng)啟用log機(jī)制時(shí)能夠加快整個(gè)作業(yè)的運(yùn)行速度，降低存儲(chǔ)開(kāi)銷(xiāo)；而任務(wù)級(jí)實(shí)現(xiàn)的優(yōu)點(diǎn)是：?jiǎn)⒂胠og機(jī)制的任務(wù)能夠加快該任務(wù)本身的運(yùn)行，減少存儲(chǔ)開(kāi)銷(xiāo)，更加靈活，當(dāng)所有任務(wù)都開(kāi)啟log機(jī)制時(shí)也能加快作業(yè)的運(yùn)行.本文的log機(jī)制是在任務(wù)級(jí)實(shí)現(xiàn)的，以任務(wù)為單位開(kāi)啟log機(jī)制.采用任務(wù)級(jí)的log機(jī)制，雖然作業(yè)的整體運(yùn)行時(shí)間要受到?jīng)]有開(kāi)啟log機(jī)制的任務(wù)運(yùn)行時(shí)間的影響，但是對(duì)于啟用log機(jī)制的任務(wù)大大減少了檢查點(diǎn)寫(xiě)入HDFS的數(shù)據(jù)量，減少了存儲(chǔ)開(kāi)銷(xiāo).而當(dāng)所有的任務(wù)都開(kāi)啟log機(jī)制時(shí)，作業(yè)的整體運(yùn)行時(shí)間也會(huì)得到很大的改善.

如果開(kāi)啟了log機(jī)制，那么在2個(gè)檢查點(diǎn)之間，每個(gè)超步都要將變化的日志寫(xiě)到遠(yuǎn)程，或者暫存在本地.這樣的話(huà)，如果這些超步累計(jì)記錄的信息大于全部動(dòng)態(tài)數(shù)據(jù)(本部分增量寫(xiě)只寫(xiě)這么多)，那么記日志就沒(méi)有優(yōu)勢(shì)可言了.對(duì)于頂點(diǎn)值發(fā)生變化的比例閾值，本文選取為檢查點(diǎn)頻率(記為c)的倒數(shù)，即1/c，此時(shí)滿(mǎn)足式(1)：

(1)

其中，P(Si)為第i超步某任務(wù)頂點(diǎn)值發(fā)生變化的比例，Si為第i個(gè)超步.此時(shí)開(kāi)啟log機(jī)制能夠保證在2個(gè)檢查點(diǎn)之間所記錄的頂點(diǎn)不會(huì)大于原來(lái)檢查點(diǎn)所記錄的頂點(diǎn)規(guī)模.另外對(duì)每個(gè)任務(wù)設(shè)置一個(gè)log機(jī)制的標(biāo)志位，用于判斷該任務(wù)的log機(jī)制是否開(kāi)啟.對(duì)于log機(jī)制開(kāi)啟條件的判斷，本文采用了一種預(yù)測(cè)式的判斷，如圖3所示.

Fig. 3 The decision of enabling log mechanism.圖3 log機(jī)制啟用判定

對(duì)提交的作業(yè)從S1開(kāi)始(S0為任務(wù)的初始化超步，不進(jìn)行記錄)對(duì)2個(gè)檢查點(diǎn)之間的每一個(gè)超步內(nèi)頂點(diǎn)值變化的頂點(diǎn)比例進(jìn)行收集，設(shè)Sk為第1個(gè)檢查點(diǎn)的超步數(shù)，一個(gè)任務(wù)在S1,S2,…,Sk滿(mǎn)足式(2)：

(2)

其中,P(Si)為第i步某任務(wù)頂點(diǎn)值變化的比例，那么該任務(wù)將會(huì)在Sk+1步開(kāi)啟log機(jī)制，該任務(wù)的log機(jī)制標(biāo)志位置為true，否則從Sk+1開(kāi)始重新收集變化的頂點(diǎn)比例.開(kāi)啟log機(jī)制后，從Sk+1繼續(xù)開(kāi)始記錄每個(gè)超步內(nèi)變化頂點(diǎn)最新的value值，到S2k時(shí)將這些頂點(diǎn)的變化記錄到HDFS的一個(gè)文件中，這就是log機(jī)制記錄的log文件.系統(tǒng)在S2k，S3k，S4k…不再記錄完整的圖頂點(diǎn)信息，而是記錄這些log信息，log的存儲(chǔ)規(guī)模遠(yuǎn)小于所有頂點(diǎn)值的存儲(chǔ)規(guī)模.

3.2 log文件生成及優(yōu)化

log機(jī)制開(kāi)啟后，如果頂點(diǎn)在參與計(jì)算之后其值發(fā)生改變，那么該頂點(diǎn)的信息將會(huì)被暫時(shí)記錄在內(nèi)存中.每記錄一個(gè)頂點(diǎn)之前首先要查找內(nèi)存中是否存在該頂點(diǎn)，如不存在直接記錄，否則記錄頂點(diǎn)的最新的值.在寫(xiě)檢查點(diǎn)時(shí)，內(nèi)存中的所有記錄將會(huì)以log文件記錄到HDFS.

開(kāi)啟log機(jī)制后，每到一個(gè)檢查點(diǎn)就會(huì)記錄一個(gè)log文件.因此，當(dāng)一個(gè)作業(yè)運(yùn)行的超步數(shù)比較多時(shí),就會(huì)在HDFS上產(chǎn)生很多l(xiāng)og文件,這些log文件會(huì)影響節(jié)點(diǎn)故障的恢復(fù).為了避免在發(fā)生節(jié)點(diǎn)故障時(shí)合并大量的log文件，任務(wù)每產(chǎn)生n個(gè)log文件(n可由配置文件讀入)就會(huì)啟動(dòng)一個(gè)線(xiàn)程在后臺(tái)合并log文件，從而減少發(fā)生節(jié)點(diǎn)故障時(shí)要合并的log文件的數(shù)量.后臺(tái)的線(xiàn)程獨(dú)立于作業(yè)的執(zhí)行，因此不會(huì)影響作業(yè)運(yùn)行時(shí)間.

當(dāng)大部分的頂點(diǎn)都發(fā)生變化時(shí)，啟用log機(jī)制的開(kāi)銷(xiāo)過(guò)大，此時(shí)檢查點(diǎn)記錄的數(shù)據(jù)量不會(huì)有明顯減少，反而會(huì)因log文件的合并增大節(jié)點(diǎn)故障的恢復(fù)開(kāi)銷(xiāo)，這時(shí)就不適合啟用log機(jī)制.采用這種策略，在啟用log機(jī)制之后只備份發(fā)生變化的動(dòng)態(tài)數(shù)據(jù)，明顯減少了檢查點(diǎn)備份的數(shù)據(jù).而在發(fā)生節(jié)點(diǎn)故障后也能根據(jù)log信息、動(dòng)態(tài)數(shù)據(jù)、靜態(tài)數(shù)據(jù)及消息進(jìn)行恢復(fù).

開(kāi)啟log機(jī)制對(duì)作業(yè)運(yùn)行的收益為

(3)

其中，p為系統(tǒng)發(fā)生節(jié)點(diǎn)故障的概率，則1-p為系統(tǒng)正常運(yùn)行至結(jié)束或發(fā)生任務(wù)故障恢復(fù)的概率;BenefitN為啟用log機(jī)制相比于沒(méi)有啟用log機(jī)制的作業(yè)正常運(yùn)行至結(jié)束或發(fā)生任務(wù)故障恢復(fù)的收益;BenefitR為啟用log機(jī)制相比于沒(méi)有啟用log機(jī)制的作業(yè)進(jìn)行節(jié)點(diǎn)故障恢復(fù)的收益.BenefitN和BenefitR可分別由式(4)和式(5)表示：

(4)

(5)

式(4)中，slog為第1次記錄log文件的超步數(shù)，(s-slog)/c+1為總共記錄的log文件的個(gè)數(shù)，CostWc k為記錄一次檢查點(diǎn)的代價(jià)，CostW(i)log為第i次記錄log的代價(jià).

式(5)中，sf表示發(fā)生節(jié)點(diǎn)故障的超步數(shù)，則(sf-slog)/c為故障任務(wù)需要讀取的log文件的數(shù)量;CostR(i)log為讀取第i個(gè)log文件的代價(jià).

為簡(jiǎn)化問(wèn)題，我們忽略在內(nèi)存中記錄變化的頂點(diǎn)信息的開(kāi)銷(xiāo)及后臺(tái)進(jìn)程對(duì)log文件的合并.由式(4)和式(5)可以看出，開(kāi)啟log機(jī)制獲得的收益和檢查點(diǎn)頻率、發(fā)生節(jié)點(diǎn)故障的超步數(shù)有密切的關(guān)系.檢查點(diǎn)頻率設(shè)置得越小，發(fā)生節(jié)點(diǎn)故障的超步數(shù)越小，開(kāi)啟log機(jī)制相對(duì)于BC-BSP的增量檢查點(diǎn)獲得的收益可能越大.

4 多級(jí)容錯(cuò)機(jī)制的故障恢復(fù)策略

4.1 任務(wù)故障的恢復(fù)

任務(wù)運(yùn)行過(guò)程中，會(huì)因?yàn)檫\(yùn)行環(huán)境的影響，例如出現(xiàn)異常、文件讀寫(xiě)錯(cuò)誤等，導(dǎo)致任務(wù)不能正常運(yùn)行.這種任務(wù)故障一般不會(huì)造成本地?cái)?shù)據(jù)的損壞(極少數(shù)的任務(wù)故障由文件的磁盤(pán)故障引起，造成文件損壞，本文忽略此種情況)，所以系統(tǒng)對(duì)于任務(wù)故障的恢復(fù)策略是直接在原來(lái)的節(jié)點(diǎn)上重新啟動(dòng)故障任務(wù)，所有任務(wù)加載檢查點(diǎn)進(jìn)行故障恢復(fù).

根據(jù)本文提出的多級(jí)容錯(cuò)處理機(jī)制的第1級(jí)容錯(cuò)處理機(jī)制的數(shù)據(jù)備份策略，本地磁盤(pán)保存了任務(wù)故障恢復(fù)所需的靜態(tài)數(shù)據(jù)和動(dòng)態(tài)數(shù)據(jù).因此，故障任務(wù)可以直接加載本地保存的靜態(tài)數(shù)據(jù)和檢查點(diǎn)時(shí)刻的動(dòng)態(tài)數(shù)據(jù)以及HDFS上備份的消息，回滾到距離故障超步最近的檢查點(diǎn)進(jìn)行故障恢復(fù).本文的第1級(jí)容錯(cuò)處理機(jī)制避免了加載HDFS保存的靜態(tài)數(shù)據(jù)和動(dòng)態(tài)數(shù)據(jù)，直接利用本地磁盤(pán)保存的靜態(tài)數(shù)據(jù)和動(dòng)態(tài)數(shù)據(jù)進(jìn)行恢復(fù)，加快了任務(wù)故障恢復(fù)時(shí)加載檢查點(diǎn)所需的時(shí)間，同時(shí)也減輕了網(wǎng)絡(luò)傳輸?shù)膲毫?

4.2 節(jié)點(diǎn)故障的恢復(fù)

節(jié)點(diǎn)故障一般是由分布式系統(tǒng)中的物理機(jī)宕機(jī)或網(wǎng)絡(luò)原因?qū)е?這種故障一般會(huì)造成故障節(jié)點(diǎn)不可用.系統(tǒng)對(duì)于節(jié)點(diǎn)故障的處理流程是：首先利用故障恢復(fù)調(diào)度機(jī)制對(duì)在這個(gè)節(jié)點(diǎn)上所有的任務(wù)進(jìn)行遷移操作，因?yàn)楣?jié)點(diǎn)發(fā)生故障就不能再使用該節(jié)點(diǎn)進(jìn)行本地恢復(fù)，故障任務(wù)將會(huì)被遷移到正常的節(jié)點(diǎn)上重啟；然后加載檢查點(diǎn)進(jìn)行恢復(fù).遷移到其他節(jié)點(diǎn)的任務(wù)由于缺少該任務(wù)以前的本地的動(dòng)態(tài)數(shù)據(jù)與靜態(tài)數(shù)據(jù)，因此系統(tǒng)通過(guò)讀取HDFS記錄的靜態(tài)數(shù)據(jù)與動(dòng)態(tài)數(shù)據(jù)及消息進(jìn)行節(jié)點(diǎn)故障的恢復(fù).

如果故障任務(wù)沒(méi)有開(kāi)啟log機(jī)制，可以利用第2級(jí)容錯(cuò)處理機(jī)制(即BC-BSP的增量檢查點(diǎn)機(jī)制)恢復(fù)策略，加載檢查點(diǎn)上的靜態(tài)數(shù)據(jù)、動(dòng)態(tài)數(shù)據(jù)及消息進(jìn)行故障恢復(fù).如果故障任務(wù)開(kāi)啟log機(jī)制，根據(jù)第3級(jí)容錯(cuò)處理機(jī)制的恢復(fù)策略，需要加載檢查點(diǎn)記錄的靜態(tài)數(shù)據(jù)、動(dòng)態(tài)數(shù)據(jù)、log文件及消息進(jìn)行節(jié)點(diǎn)故障恢復(fù).第2級(jí)容錯(cuò)處理機(jī)制已在2.2節(jié)詳細(xì)介紹，這里不再贅述.

第3級(jí)容錯(cuò)處理機(jī)制的恢復(fù)策略具體為：按照l(shuí)og文件生成的先后順序，首先讀取最晚生成的log文件的每一條記錄到內(nèi)存中；然后依次讀取較早生成的log文件，較早記錄的log文件中的頂點(diǎn)ID如在內(nèi)存中已記錄就無(wú)需再記錄，而較早記錄的log文件中的頂點(diǎn)在內(nèi)存中不存在時(shí)便記錄此頂點(diǎn)信息.掃描完所有的log文件之后，再讀取記錄有全圖信息的檢查點(diǎn)記錄(可能存在增量檢查點(diǎn)，也可能只存在第1個(gè)檢查點(diǎn)記錄)，將內(nèi)存中的記錄與檢查點(diǎn)按照上述方法再次合并，最終生成故障發(fā)生前最近的檢查點(diǎn)時(shí)刻的完整動(dòng)態(tài)數(shù)據(jù).該策略完整描述如算法3所示.

算法3.readLogCheckPoint(s，ck).

輸入：記錄第1個(gè)log文件的超步數(shù)s、檢查點(diǎn)頻率ck；

輸出：圖數(shù)據(jù)對(duì)象graphData.

① /*讀取未合并log文件集合*/

② For each log filelfrinNr/*Nr:未合并log文件集合*/

③Foreach頂點(diǎn)vinlfr

④ 將v放入c /*c:值發(fā)生變化的頂點(diǎn)集合*/

⑤ End For

⑥ End For

⑦ /*讀取已合并log文件集*/

⑧ For each log filelfminNm/*Nm:已合并log文件集合*/

⑨Foreach頂點(diǎn)vinlfm

⑩If不存在v

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

5.1 數(shù)據(jù)集與實(shí)驗(yàn)設(shè)置

本文使用2個(gè)真實(shí)圖數(shù)據(jù)集進(jìn)行實(shí)驗(yàn)，包括Wiki[12]和USA-Road[13]，具體描述如表1所示.測(cè)試使用的應(yīng)用包括計(jì)算圖的連通分量(CC)和單源最短路徑(SSSP).

Table 1 Description of Real-World Graphs

本文在BC-BSP系統(tǒng)上實(shí)現(xiàn)了多級(jí)容錯(cuò)處理機(jī)制.本文實(shí)驗(yàn)的對(duì)比分析包括：第1級(jí)容錯(cuò)處理機(jī)制與BC-BSP的增量檢查點(diǎn)機(jī)制的對(duì)比，即任務(wù)故障恢復(fù)時(shí)加載HDFS與加載本地?cái)?shù)據(jù)的對(duì)比；不同寫(xiě)檢查點(diǎn)頻率下開(kāi)啟log機(jī)制與關(guān)閉log機(jī)制的作業(yè)運(yùn)行時(shí)間、寫(xiě)檢查點(diǎn)IO開(kāi)銷(xiāo)(不包括消息)的對(duì)比；第3級(jí)容錯(cuò)處理機(jī)制與BC-BSP增量檢查點(diǎn)的對(duì)比，不同寫(xiě)檢查點(diǎn)頻率下節(jié)點(diǎn)故障的恢復(fù)；頂點(diǎn)值變化比例的閾值對(duì)作業(yè)運(yùn)行的影響.第2級(jí)容錯(cuò)處理機(jī)制即為原BC-BSP系統(tǒng)節(jié)點(diǎn)故障的恢復(fù)機(jī)制.實(shí)驗(yàn)所用集群由15個(gè)節(jié)點(diǎn)構(gòu)成，且由1臺(tái)Gigabit以太網(wǎng)交換機(jī)連接，每個(gè)計(jì)算節(jié)點(diǎn)配置酷睿i3-2100雙核處理器、8 GB內(nèi)存、1TB的7200RPM硬盤(pán).每個(gè)節(jié)點(diǎn)最大任務(wù)槽數(shù)設(shè)為2，測(cè)試時(shí)啟動(dòng)10個(gè)任務(wù)，多余的節(jié)點(diǎn)是為了發(fā)生節(jié)點(diǎn)故障時(shí)有可用的節(jié)點(diǎn)進(jìn)行故障遷移.節(jié)點(diǎn)的心跳間隔設(shè)為1 s，心跳超時(shí)時(shí)間設(shè)為3 s.測(cè)試的參數(shù)為檢查點(diǎn)頻率和故障超步數(shù)，測(cè)試的指標(biāo)為作業(yè)運(yùn)行時(shí)間和寫(xiě)檢查點(diǎn)IO開(kāi)銷(xiāo).

5.2 任務(wù)故障恢復(fù)

我們?cè)?個(gè)真實(shí)數(shù)據(jù)集上使用SSSP和CC測(cè)試了BC-BSP增量檢查點(diǎn)機(jī)制加載HDFS與多級(jí)容錯(cuò)處理機(jī)制的第1級(jí)容錯(cuò)處理機(jī)制加載本地磁盤(pán)進(jìn)行任務(wù)故障恢復(fù)的時(shí)間.這里我們?cè)O(shè)置檢查點(diǎn)頻率為6，運(yùn)行至第8步發(fā)生任務(wù)故障，共運(yùn)行10個(gè)超步.

圖4為不同的應(yīng)用分別在BC-BSP的增量檢查點(diǎn)機(jī)制與本文的多級(jí)容錯(cuò)機(jī)制的第1級(jí)容錯(cuò)機(jī)制下進(jìn)行任務(wù)故障恢復(fù)的運(yùn)行時(shí)間，圖5統(tǒng)計(jì)了平均每個(gè)任務(wù)在進(jìn)行任務(wù)故障恢復(fù)時(shí)加載檢查點(diǎn)所花費(fèi)的時(shí)間.

Fig. 4 The recovery time of task failure.圖4 任務(wù)故障恢復(fù)時(shí)間

Fig. 5 The checkpoint time of task failure recovery.圖5 任務(wù)故障恢復(fù)加載檢查點(diǎn)時(shí)間

綜合圖4和圖5的實(shí)驗(yàn)結(jié)果可以發(fā)現(xiàn)，對(duì)于不同應(yīng)用(SSSP和CC)，加載本地?cái)?shù)據(jù)的時(shí)間比加載HDFS的時(shí)間快了1倍多，作業(yè)恢復(fù)運(yùn)行的總時(shí)間也有所改善.此外，由于加載本地?cái)?shù)據(jù)不需要網(wǎng)絡(luò)傳輸，因此也降低了網(wǎng)絡(luò)傳輸?shù)拈_(kāi)銷(xiāo).本節(jié)的實(shí)驗(yàn)證明了本文多級(jí)容錯(cuò)處理機(jī)制的第1級(jí)容錯(cuò)處理機(jī)制的高效性.

BC-BSP進(jìn)入迭代計(jì)算階段，每個(gè)超步結(jié)束后將動(dòng)態(tài)變化的頂點(diǎn)數(shù)據(jù)寫(xiě)回本地磁盤(pán)，不發(fā)生變化的靜態(tài)數(shù)據(jù)只在需要處理時(shí)再?gòu)谋镜卮疟P(pán)讀入內(nèi)存，而在處理結(jié)束時(shí)并不寫(xiě)回磁盤(pán)，因?yàn)樗鼪](méi)有變化.任務(wù)故障的恢復(fù)只需額外備份檢查點(diǎn)寫(xiě)到磁盤(pán)上的動(dòng)態(tài)數(shù)據(jù)，所以存儲(chǔ)代價(jià)為一步的動(dòng)態(tài)數(shù)據(jù)的大小.經(jīng)實(shí)驗(yàn)測(cè)得，對(duì)于2種應(yīng)用在USA-Road數(shù)據(jù)集上，每臺(tái)機(jī)器的存儲(chǔ)代價(jià)均為21 MB，而Wiki數(shù)據(jù)集的存儲(chǔ)代價(jià)均為每臺(tái)機(jī)器5MB.因此，存儲(chǔ)代價(jià)很低.

5.3 log機(jī)制對(duì)正常運(yùn)行的作業(yè)的影響

在2個(gè)真實(shí)數(shù)據(jù)集上，使用SSSP測(cè)試了log機(jī)制，以證明開(kāi)啟log機(jī)制能夠加速作業(yè)正常運(yùn)行.

圖6給出了SSSP在數(shù)據(jù)集Wiki和USA-Road上以不同的檢查點(diǎn)頻率正常運(yùn)行40個(gè)超步時(shí)，BC-BSP的增量檢查點(diǎn)機(jī)制與log機(jī)制的運(yùn)行時(shí)間的對(duì)比.

Fig. 6 Running time of job against the checkpoint frequency.圖6 不同檢查點(diǎn)頻率作業(yè)運(yùn)行時(shí)間

Fig. 7 IO cost of backup against the checkpoint frequency.圖7 不同檢查點(diǎn)頻率下備份的IO開(kāi)銷(xiāo)

由圖6可以看出，開(kāi)啟log機(jī)制后的作業(yè)運(yùn)行時(shí)間明顯減少，而寫(xiě)檢查點(diǎn)頻率設(shè)置越小，log機(jī)制對(duì)于作業(yè)正常運(yùn)行時(shí)間的收益越大.因?yàn)椋_(kāi)啟log機(jī)制的任務(wù)在寫(xiě)檢查點(diǎn)時(shí)備份的數(shù)據(jù)量比BC-BSP的增量檢查點(diǎn)機(jī)制要少得多，在記錄多個(gè)檢查點(diǎn)后，log機(jī)制明顯地縮短了作業(yè)正常運(yùn)行的時(shí)間.

圖7給出了SSSP在數(shù)據(jù)集Wiki和USA-Road上以不同的檢查點(diǎn)頻率正常運(yùn)行40個(gè)超步時(shí)，BC-BSP的增量檢查點(diǎn)機(jī)制與log機(jī)制的備份IO開(kāi)銷(xiāo)對(duì)比.由圖7可以看出，啟用log機(jī)制后備份的IO開(kāi)銷(xiāo)比BC-BSP的增量檢查點(diǎn)機(jī)制的要小，特別是對(duì)USA-Road這種平均出度比較小的數(shù)據(jù)集效果更加顯著.這是因?yàn)镾SSP在USA-Road啟用log機(jī)制的超步比在Wiki上早很多，因此SSSP在USA-Road上的IO收益要遠(yuǎn)高于在Wiki上的IO收益，在時(shí)間上的收益也高于Wiki.本節(jié)從時(shí)間和備份的IO開(kāi)銷(xiāo)的角度來(lái)對(duì)比BC-BSP的增量檢查點(diǎn)機(jī)制與log機(jī)制，實(shí)驗(yàn)結(jié)果論證了本文的log機(jī)制提高了作業(yè)正常運(yùn)行的效率.

5.4 log機(jī)制對(duì)節(jié)點(diǎn)故障恢復(fù)的影響

我們?cè)?個(gè)真實(shí)數(shù)據(jù)集使用SSSP測(cè)試了log機(jī)制對(duì)于節(jié)點(diǎn)故障恢復(fù)的影響，為了說(shuō)明故障超步數(shù)和檢查點(diǎn)頻率對(duì)節(jié)點(diǎn)故障恢復(fù)的影響，我們?cè)赪iki數(shù)據(jù)集上進(jìn)行了測(cè)試.

圖8和圖9分別給出了以不同的檢查點(diǎn)頻率在第17步與第33步制造節(jié)點(diǎn)故障時(shí)運(yùn)行40個(gè)超步log機(jī)制對(duì)于節(jié)點(diǎn)故障恢復(fù)時(shí)間的影響.

Fig. 8 Recovery time of job against the checkpoint frequency.圖8 不同檢查點(diǎn)頻率作業(yè)恢復(fù)時(shí)間

Fig. 9 Recovery time of job against the checkpoint frequency.圖9 不同檢查點(diǎn)頻率作業(yè)恢復(fù)時(shí)間

對(duì)比圖8和圖9可以發(fā)現(xiàn)，故障步數(shù)越大，恢復(fù)時(shí)需要讀取的log文件內(nèi)容可能越多，合并log所花費(fèi)的時(shí)間開(kāi)銷(xiāo)也有所增大，但是恢復(fù)的總時(shí)間仍小于沒(méi)有開(kāi)啟log機(jī)制的總時(shí)間.這是因?yàn)椋趩⒂胠og機(jī)制的情況下，寫(xiě)檢查點(diǎn)的時(shí)間開(kāi)銷(xiāo)減少了,節(jié)省的時(shí)間足以抵消讀取log文件所花費(fèi)的時(shí)間.因此開(kāi)啟log機(jī)制在一定程度上加速了節(jié)點(diǎn)故障的恢復(fù)過(guò)程.本節(jié)實(shí)驗(yàn)說(shuō)明第3級(jí)容錯(cuò)處理機(jī)制加速了節(jié)點(diǎn)故障的恢復(fù).

5.5 log啟動(dòng)閾值對(duì)log機(jī)制的影響

我們使用SSSP和CC在Wiki上測(cè)試不同的log啟動(dòng)閾值對(duì)于log機(jī)制的影響，以驗(yàn)證3.1節(jié)理論分析.本節(jié)實(shí)驗(yàn)中，閾值的定義為值發(fā)生變化的頂點(diǎn)占所有頂點(diǎn)的比例，而檢查點(diǎn)頻率設(shè)置為5，運(yùn)行40個(gè)超步.

圖10和圖11分別給出了在不同閾值下作業(yè)的運(yùn)行時(shí)間與備份的IO開(kāi)銷(xiāo).該閾值的選取要對(duì)作業(yè)運(yùn)行時(shí)間和寫(xiě)檢查點(diǎn)的IO開(kāi)銷(xiāo)優(yōu)化相對(duì)多.因?yàn)殚撝颠x取過(guò)大可能會(huì)造成在內(nèi)存中記錄的log信息過(guò)多，從而導(dǎo)致內(nèi)存開(kāi)銷(xiāo)過(guò)大；圖11可以看出，閾值選取為10%時(shí)，啟用log機(jī)制比較晚，導(dǎo)致備份的IO開(kāi)銷(xiāo)比較大.通過(guò)權(quán)衡作業(yè)的運(yùn)行時(shí)間和寫(xiě)檢率的倒數(shù))是比較合適的.當(dāng)閾值為20%時(shí)，寫(xiě)檢查點(diǎn)的IO相對(duì)較小，作業(yè)的運(yùn)行時(shí)間也和其他3種閾值下的作業(yè)運(yùn)行時(shí)間相當(dāng).這說(shuō)明了3.1節(jié)中對(duì)這個(gè)閾值的推導(dǎo)是正確的.

Fig. 10 Running time of job against the threshold of starting the log mechanism.圖10 不同log機(jī)制啟動(dòng)閾值下作業(yè)運(yùn)行時(shí)間

Fig. 11 IO cost of backup against the threshold of starting the log mechanism.圖11 不同log機(jī)制啟動(dòng)閾值下備份的IO開(kāi)銷(xiāo)

6 結(jié) 論

本文提出了多級(jí)容錯(cuò)處理機(jī)制，通過(guò)在2個(gè)真實(shí)數(shù)據(jù)集上大量的對(duì)比實(shí)驗(yàn)證明了多級(jí)容錯(cuò)機(jī)制的高效性與正確性.第1級(jí)容錯(cuò)處理機(jī)制直接利用本地保存的動(dòng)態(tài)數(shù)據(jù)、靜態(tài)數(shù)據(jù)及HDFS上的消息進(jìn)行恢復(fù)，避免了加載HDFS上動(dòng)態(tài)數(shù)據(jù)、靜態(tài)數(shù)據(jù)從而加快了其恢復(fù)過(guò)程.第3級(jí)容錯(cuò)處理機(jī)制對(duì)于頂點(diǎn)值變化比例較低的應(yīng)用，例如SSSP和CC，通過(guò)log記錄變化的頂點(diǎn)信息而極大減少了傳統(tǒng)檢查點(diǎn)機(jī)制所記錄的數(shù)據(jù)量和存儲(chǔ)開(kāi)銷(xiāo)(實(shí)驗(yàn)中也發(fā)現(xiàn)，對(duì)于每個(gè)超步頂點(diǎn)變化比例較高的應(yīng)用，例如PageRank，意義不大).系統(tǒng)在所有任務(wù)都啟用log機(jī)制后，整個(gè)作業(yè)的運(yùn)行時(shí)間明顯減少.通過(guò)log信息進(jìn)行節(jié)點(diǎn)故障的恢復(fù)，在節(jié)點(diǎn)恢復(fù)過(guò)程中雖然引入了合并log的過(guò)程，但由于作業(yè)運(yùn)行過(guò)程中開(kāi)啟了log機(jī)制，作業(yè)的整體運(yùn)行時(shí)間在一定程度上仍有所降低.

下一步的工作將探索日志合并頻率對(duì)log機(jī)制的影響，即它的改變對(duì)于節(jié)點(diǎn)故障恢復(fù)時(shí)間的影響.通過(guò)大量實(shí)驗(yàn)找出一個(gè)最合適節(jié)點(diǎn)故障恢復(fù)的日志合并頻率.

[1]The Apache Software Foundation. Introduction to Giraph[EB/OL]. [2015-05-25]. http://giraph.apache.org/intro.html

[2]Bao Yubin, Wang Zhigang, Yu Gu, et al. BC-BSP: A BSP-based parallel iterative processing system for big data on cloud architecture[C] //Proc of the 1st Int DASFAA Workshop on Big Data Management and Analytics. Berlin: Springer, 2013: 31-45

[3]Malewicz G, Austern M H, Bik A J C, et al. Pregel: A system for large-scale graph processing[C] //Proc of the 2010 ACM SIGMOD Int Conf on Management of Data. New York: ACM, 2010: 135-146

[4]Shen Y, Chen G, Jagadish H V, et al. Fast failure recovery in distributed graph processing systems[J]. Proceedings of the VLDB Endowment, 2014, 8(4): 437-448

[5]Low Y, Gonzalez J E, Kyrola A, et al. GraphLab: A new framework for parallel machine learning[J/OL]. 2014[2015-05-25]. http://arxiv.org/abs/1408.2041

[6]Gonzalez J E, Low Y, Gu H, et al. PowerGraph: Distributed graph-parallel computation on natural graphs[C] //Proc of the 10th USENIX Conf on Operating Systems Design and Implementation. Berkeley, CA: USENIX Association, 2012: 17-30

[7]Salihoglu S, Widom J. GPS: A graph processing system[C] //Proc of the 25th Int Conf on Scientific and Statistical Database Management. New York: ACM, 2013: 22

[8]Khayyat Z, Awara K, Alonazi A, et al. Mizan: A system for dynamic load balancing in large-scale graph processing [C] //Proc of the 8th ACM European Conf on Computer Systems. New York: ACM, 2013: 169-182

[9]Xin R S, Gonzalez J E, Franklin M J, et al. GraphX: A resilient distributed graph system on spark[C] //Proc of the 1st Int Workshop on Graph Data Management Experiences and Systems. New York: ACM, 2013: 1-6

[10]Zaharia M, Chowdhury M, Das T, et al. Resilient distributed datasets: A fault-tolerant abstraction for in-memory cluster computing[C] //Proc of the 9th USENIX Conf on Networked Systems Design and Implementation. Berkeley, CA: USENIX Association, 2012: 141-146

[11]Yi Huizhan, Wang Feng, Zuo Ke, et al. Asynchronous checkpoint/restart based on memory buffer[J]. Journal of Computer Research and Development, 2015, 52(6): 1229-1239 (in Chinese)

(易會(huì)戰(zhàn), 王鋒, 左克, 等. 基于內(nèi)存緩存的異步檢查點(diǎn)容錯(cuò)技術(shù)[J]. 計(jì)算機(jī)研究與發(fā)展, 2015, 52(6): 1229-1239)

[12]Wikipedia. Using the Wikipedia Link[EB/OL]. [2015-05-25]. http://haselgrove.id.au/wikipedia.htm[13]Sapienza University of Rome. Using the USA-Road Link[EB/OL]. [2015-05-25]. http://www.dis. uniroma1.it/challenge9/download.shtml

Bi Yahui, born in 1990. Master candidate at the College of Computer Science and Engineering, Northeastern University. His main research interests include cloud computing and graph management, etc.

Jiang Suyang, born in 1991. Master candidate at the College of Computer Science and Engineering, Northeastern University. Her main research interests include cloud computing and graph management, etc.

Wang Zhigang, born in 1987. PhD candidate at the College of Computer Science and Engineering, Northeastern University. His main research interests include cloud computing and graph data mining, etc.

Leng Fangling, born in 1978. Received her PhD degree in computer software and theory from Northeastern University in 2008. Lecturer at Northeastern University. Member of China Computer Federation. Her main research interests include data warehouse and online analytical processing (OLAP), etc.

Bao Yubin, born in 1968. Received his PhD degree in computer software and theory from Northeastern University in 2003. Professor at Northeastern University. Senior member of China Computer Federation. His main research interests include data warehouse, online analytical processing (OLAP), cloud computing and data intensive computing, etc.

Yu Ge, born in 1962. Received his PhD degree in computer science from Kyushu University of Japan in 1996. Professor and PhD supervisor at Northeastern University. His main research interests include database theory and technology, distributed system, parallel computing and cloud computing, etc.

Qian Ling, born in 1972. Received his PhD degree of engineering at the Department of Computer Science and Technology, Tsinghua University, in 2001. He joined Bell Labs Research China in 2001. He worked on IPv6 edge router, voice messaging, voip, instant messaging, LBS, mobile application and other related projects. In 2008, he joined China Mobile Research Institute and worked on mobiles ads, big data and cloud computing projects.

A Multi-Level Fault Tolerance Mechanism for Disk-Resident Pregel-Like Systems

Bi Yahui1, Jiang Suyang1, Wang Zhigang1, Leng Fangling1, Bao Yubin1, Yu Ge1, and Qian Ling2

1(CollegeofComputerScienceandEngineering,NortheasternUniversity,Shenyang110819)2(ChinaMobile(Suzhou)SoftwareTechnologyCo,Ltd,Suzhou,Jiangsu215163)

The BSP-based distributed frameworks, such as Pregel, are becoming a powerful tool for handling large-scale graphs, especially for applications with iterative computing frequently. Distributed systems can guarantee a flexible processing capacity by adding computing nodes, however, they also increase the probability of failures. Therefore, an efficient fault-tolerance mechanism is essential. Existing work mainly focuses on the checkpoint policy, including backup and recovery. The former usually backups all graph data, which leads to the cost of writing redundant data since some data are static during iterations. The latter always loads backup data from remote machines to recovery iterations, ignoring the usage of data in the local disk in special scenarios, which incurs network costs. It proposes a multi-level fault tolerant mechanism, which distinguishes failures into computing task failures and node failures, and then designs different strategies for backup and recovery. For the latter, considering that the volume of data involved in computation varies with iterations, a complete backup policy and an adaptive log-based policy are presented to reduce the cost of writing redundant data. After that, at the stages of recovery, we utilize the local graph data and the remote message data to handle the recovery for task failures, but the remote data are used for node failures. Finally, extensive experiments on real datasets validate the efficiency of our solutions.

fault tolerance;large-scale graph; iterative computing; BSP model; checkpoint

2015-06-30；

2015-10-29

國(guó)家自然科學(xué)基金重點(diǎn)項(xiàng)目(61433008);國(guó)家自然科學(xué)基金項(xiàng)目(61173028,61272179);中央高校基本科研業(yè)務(wù)費(fèi)專(zhuān)項(xiàng)基金項(xiàng)目(N100704001);教育部-中國(guó)移動(dòng)科研基金項(xiàng)目(MCM20125021)

TP311. 13

This work was supported by the Key Program of the National Natural Science Foundation of China (61433008), the National Natural Science Foundation of China (61173028,61272179), the Fundamental Research Funds for the Central Universities (N100704001), and Chinese Ministry of Education-China Mobile Communications Corporation Research Funds (MCM20125021).