• <tr id="yyy80"></tr>
  • <sup id="yyy80"></sup>
  • <tfoot id="yyy80"><noscript id="yyy80"></noscript></tfoot>
  • 99热精品在线国产_美女午夜性视频免费_国产精品国产高清国产av_av欧美777_自拍偷自拍亚洲精品老妇_亚洲熟女精品中文字幕_www日本黄色视频网_国产精品野战在线观看 ?

    基于RDD關(guān)鍵度的Spark檢查點(diǎn)管理策略

    2017-12-16 05:09:32英昌甜王維慶錢育蓉
    計(jì)算機(jī)研究與發(fā)展 2017年12期
    關(guān)鍵詞:檢查點(diǎn)內(nèi)存分區(qū)

    英昌甜 于 炯 卞 琛 王維慶 魯 亮 錢育蓉

    1(新疆大學(xué)電氣工程博士后科研流動(dòng)站 烏魯木齊 830046) 2(新疆大學(xué)軟件學(xué)院 烏魯木齊 830008) 3((新疆大學(xué)電氣工程學(xué)院 烏魯木齊 830046)

    基于RDD關(guān)鍵度的Spark檢查點(diǎn)管理策略

    英昌甜1,2于 炯2卞 琛2王維慶1,3魯 亮2錢育蓉2

    1(新疆大學(xué)電氣工程博士后科研流動(dòng)站 烏魯木齊 830046)2(新疆大學(xué)軟件學(xué)院 烏魯木齊 830008)3((新疆大學(xué)電氣工程學(xué)院 烏魯木齊 830046)

    (yingct@xju.edu.com)

    Spark默認(rèn)容錯(cuò)機(jī)制由程序員設(shè)置檢查點(diǎn),并利用彈性分布式數(shù)據(jù)集(resilient distributed dataset, RDD)的血統(tǒng)(lineage)進(jìn)行計(jì)算.在應(yīng)用程序復(fù)雜度高、迭代次數(shù)多以及數(shù)據(jù)量較大時(shí),恢復(fù)過(guò)程需要耗費(fèi)大量的計(jì)算開銷.同時(shí),在執(zhí)行恢復(fù)任務(wù)時(shí),僅考慮數(shù)據(jù)本地性選擇節(jié)點(diǎn),并未考慮節(jié)點(diǎn)的計(jì)算能力,這都會(huì)導(dǎo)致恢復(fù)時(shí)間增加,無(wú)法最大化發(fā)揮集群的性能.因此,在建立Spark執(zhí)行模型、檢查點(diǎn)模型和RDD關(guān)鍵度模型的基礎(chǔ)上,提出一種基于關(guān)鍵度的檢查點(diǎn)管理(criticality checkpoint management, CCM)策略,其中包括檢查點(diǎn)設(shè)置算法、失效恢復(fù)算法和清理算法.其中檢查點(diǎn)設(shè)置算法通過(guò)分析作業(yè)中RDD的屬性以及對(duì)作業(yè)恢復(fù)時(shí)間的影響,選擇關(guān)鍵度大的RDD作為檢查點(diǎn)存儲(chǔ);恢復(fù)算法根據(jù)各節(jié)點(diǎn)的計(jì)算能力做出決策,選擇合適的節(jié)點(diǎn)執(zhí)行恢復(fù)任務(wù);清理算法在磁盤空間不足時(shí),清除關(guān)鍵度較低的檢查點(diǎn).實(shí)驗(yàn)結(jié)果表明:該策略在略增加執(zhí)行時(shí)間的情況下,能夠選擇有備份價(jià)值的RDD作為檢查點(diǎn),在節(jié)點(diǎn)失效時(shí)能夠有效地降低恢復(fù)開銷,提高節(jié)點(diǎn)的磁盤有效利用率.

    內(nèi)存計(jì)算;Spark;檢查點(diǎn)管理;失效恢復(fù);RDD屬性

    近年來(lái),隨著互聯(lián)網(wǎng)的快速發(fā)展,特別是云計(jì)算的普及,全球數(shù)據(jù)量以每年約50%的速度遞增,大數(shù)據(jù)[1-3]日益受到人們的重視.2015年9月,國(guó)務(wù)院印發(fā)《促進(jìn)大數(shù)據(jù)發(fā)展行動(dòng)綱要》,確定了大數(shù)據(jù)發(fā)展的國(guó)家頂層設(shè)計(jì),大數(shù)據(jù)與各行各業(yè)的結(jié)合已是行業(yè)未來(lái)發(fā)展的必然趨勢(shì).隨著大數(shù)據(jù)時(shí)代的到來(lái),數(shù)據(jù)每天都在急劇快速膨脹,發(fā)掘這些數(shù)據(jù)的價(jià)值,需要一種高效而穩(wěn)定的分布式計(jì)算框架和模型.在分布式計(jì)算框架中,Apache Spark[4-6]由于其基于內(nèi)存的高性能計(jì)算模式以及豐富靈活的編程接口,得到了廣泛的支持和應(yīng)用,大有逐漸取代Hadoop MapReduce成為新一代大數(shù)據(jù)計(jì)算引擎的趨勢(shì).

    Spark采用了基于分布式共享內(nèi)存的彈性分布式數(shù)據(jù)集(resilient distributed datasets, RDD)[7]作為數(shù)據(jù)結(jié)構(gòu),RDD是Spark對(duì)分布式內(nèi)存的抽象,具有可重構(gòu)性、不變性、分區(qū)局部性以及可序列化等特性[8-10].作為Apache頂級(jí)的開源項(xiàng)目,Spark基于自身的核心部分,在迭代計(jì)算、交互式查詢計(jì)算以及批量流計(jì)算等方面,開發(fā)了適應(yīng)大數(shù)據(jù)處理的多種場(chǎng)景生態(tài)組件,如SQL處理引擎Spark SQL和Shark、流式處理引擎Spark Streaming、機(jī)器學(xué)習(xí)系統(tǒng)MLLib、圖計(jì)算框架GraphX、統(tǒng)計(jì)分析工具SparkR以及分布式內(nèi)存文件系統(tǒng)Tachyon等.

    Spark是基于MapReduce算法實(shí)現(xiàn)的分布式計(jì)算框架,因此具有MapReduce的優(yōu)點(diǎn)[11-12].同時(shí)Spark在很多方面都彌補(bǔ)了MapReduce的不足,比MapReduce的通用性更好、迭代運(yùn)算效率更高、作業(yè)延遲更低.Spark的主要優(yōu)勢(shì)包括:提供了一套支持DAG圖的分布式并行計(jì)算的編程框架,減少多次計(jì)算之間中間結(jié)果寫到HDFS的開銷;提供Cache機(jī)制來(lái)支持需要反復(fù)迭代計(jì)算或者多次數(shù)據(jù)共享,減少數(shù)據(jù)讀取的I/O開銷;使用多線程池模型來(lái)減少任務(wù)啟動(dòng)開銷,shuffle過(guò)程中避免不必要的sort操作以及減少磁盤I/O操作;具有更廣泛的數(shù)據(jù)集操作類型.

    在Spark任務(wù)的執(zhí)行過(guò)程中,首先將HDFS中存儲(chǔ)的數(shù)據(jù)作為數(shù)據(jù)源加載到RDD中進(jìn)行處理,再通過(guò)一系列的操作生成計(jì)算結(jié)果,每次操作所產(chǎn)生的中間結(jié)果都以RDD的形式存儲(chǔ)在內(nèi)存中.Spark通過(guò)數(shù)據(jù)集的血統(tǒng)(lineage)實(shí)現(xiàn)容錯(cuò),當(dāng)節(jié)點(diǎn)失效導(dǎo)致數(shù)據(jù)丟失時(shí),Spark可以根據(jù)血統(tǒng)重新計(jì)算,以實(shí)現(xiàn)丟失數(shù)據(jù)的自動(dòng)重建.為了避免錯(cuò)誤恢復(fù)的代價(jià)與運(yùn)行時(shí)間成正比增長(zhǎng),Spark提供了檢查點(diǎn)(checkpoint)[13]功能,通過(guò)設(shè)置檢查點(diǎn)來(lái)記錄中間狀態(tài),避免從頭開始計(jì)算的漫長(zhǎng)恢復(fù).

    然而,Spark集群默認(rèn)的容錯(cuò)機(jī)制將檢查點(diǎn)選擇和設(shè)置的權(quán)利交給編程人員,而程序員往往根據(jù)經(jīng)驗(yàn)進(jìn)行選擇,使結(jié)果充滿不確定性.數(shù)據(jù)丟失后恢復(fù)的效果好壞和效率高低,會(huì)隨著程序員的水平不同而出現(xiàn)巨大差異.錯(cuò)誤的檢查點(diǎn)策略不僅會(huì)導(dǎo)致程序變慢,恢復(fù)效率降低,甚至?xí)速M(fèi)持久化存儲(chǔ)(磁盤/固態(tài)硬盤等)的空間,影響其他作業(yè)執(zhí)行效率.此外,由于現(xiàn)有分布式框架中的檢查點(diǎn)策略,往往根據(jù)時(shí)間周期定時(shí)設(shè)置,并未對(duì)具體作業(yè)進(jìn)行分析,考慮影響作業(yè)恢復(fù)效率的重要因素,因此選取的檢查點(diǎn)并不一定有效,反而可能影響整體系統(tǒng)的計(jì)算和容錯(cuò)性能.

    為解決這一問(wèn)題,本文主要做了3方面工作:

    1) 對(duì)內(nèi)存計(jì)算框架的作業(yè)執(zhí)行機(jī)制進(jìn)行分析,建立執(zhí)行效率模型,給出了RDD計(jì)算代價(jià)和任務(wù)執(zhí)行時(shí)間的定義.

    2) 通過(guò)分析RDD的恢復(fù)過(guò)程,建立了檢查點(diǎn)恢復(fù)模型,給出了RDD關(guān)鍵度、失效恢復(fù)比的定義,并證明這些定義與任務(wù)恢復(fù)效率的關(guān)系,為算法設(shè)計(jì)提供基礎(chǔ)模型.

    3) 在相關(guān)模型定義和證明的基礎(chǔ)上,提出了檢查點(diǎn)管理策略的問(wèn)題定義,以此作為算法設(shè)計(jì)的主要依據(jù).

    1 相關(guān)工作

    檢查點(diǎn)/恢復(fù)策略是分布式計(jì)算廣泛使用的容錯(cuò)技術(shù),在國(guó)內(nèi)外都有較多的相關(guān)研究[14-19].傳統(tǒng)的檢查點(diǎn)策略主要分為3類:應(yīng)用級(jí)實(shí)現(xiàn)、用戶級(jí)實(shí)現(xiàn)和系統(tǒng)級(jí)實(shí)現(xiàn).

    1) 應(yīng)用級(jí)實(shí)現(xiàn).由編程人員或自動(dòng)程序?qū)z查點(diǎn)代碼與應(yīng)用程序代碼整合,檢查點(diǎn)活動(dòng)由應(yīng)用程序自動(dòng)執(zhí)行.將檢查點(diǎn)存儲(chǔ)到持久化存儲(chǔ)中,并且當(dāng)節(jié)點(diǎn)故障后,從檢查點(diǎn)重啟.這種方法的挑戰(zhàn)在于需要編程人員對(duì)設(shè)置檢查點(diǎn)的應(yīng)用有透徹的了解.

    2) 用戶級(jí)實(shí)現(xiàn).需要用戶級(jí)的數(shù)據(jù)庫(kù)來(lái)設(shè)置檢查點(diǎn),并且應(yīng)用程序與數(shù)據(jù)庫(kù)相鏈接,這種方法對(duì)用戶不透明,因?yàn)閼?yīng)用的修改、編譯都鏈接到檢查點(diǎn)數(shù)據(jù)庫(kù).

    3) 系統(tǒng)級(jí)實(shí)現(xiàn).檢查點(diǎn)/恢復(fù)策略可以實(shí)現(xiàn)在系統(tǒng)級(jí)別,包括OS內(nèi)核或硬件.當(dāng)實(shí)現(xiàn)在系統(tǒng)級(jí)別,對(duì)用戶通常是透明的,并且對(duì)應(yīng)用程序不需要修改.

    另外一些研究成果主要考慮分布式環(huán)境下并行計(jì)算框架的失效恢復(fù)策略.Storm[20]是一個(gè)實(shí)時(shí)流式處理平臺(tái),運(yùn)行在Java虛擬機(jī)之上;使用Clojure和Java編寫,并且支持多語(yǔ)言編程;它依靠Nimbus在集群分發(fā)代碼,并將任務(wù)分配給工作節(jié)點(diǎn);由Zookeeper為其提供了容錯(cuò),并與Nimbus協(xié)調(diào)重新分配任務(wù)到其他可用節(jié)點(diǎn);然而,沒(méi)有設(shè)置檢查點(diǎn)機(jī)制.Apache S4[21]是一個(gè)開源的流處理平臺(tái),利用Zookeeper提供容錯(cuò).與Storm不同,它提供了檢查點(diǎn)機(jī)制.在一個(gè)節(jié)點(diǎn)出現(xiàn)故障時(shí),故障轉(zhuǎn)移機(jī)制重新啟動(dòng)一個(gè)新節(jié)點(diǎn)進(jìn)行任務(wù)恢復(fù).為了減少延遲,檢查點(diǎn)是異步的,這意味著每個(gè)獨(dú)立節(jié)點(diǎn)執(zhí)行檢查點(diǎn),沒(méi)有全局一致性.文獻(xiàn)[22-23]提出了一種基于內(nèi)存的分布式文件系統(tǒng)Tachyon,兼容包括Spark在內(nèi)的多種計(jì)算框架.Tachyon本質(zhì)上是將內(nèi)存存儲(chǔ)功能從Spark或Yarn中分離出來(lái),使上層計(jì)算框架實(shí)現(xiàn)更高的執(zhí)行效率.在Tachyon的實(shí)現(xiàn)中,文件采用兼容HDFS的Block方式進(jìn)行管理,使用不同的存儲(chǔ)媒介對(duì)數(shù)據(jù)分層次緩存,但檢查點(diǎn)算法僅保存當(dāng)前最新生成的RDD,與任務(wù)和RDD的特性無(wú)關(guān).文獻(xiàn)[24]提出了內(nèi)存文件系統(tǒng)RAMCloud,將內(nèi)存作為文件的存儲(chǔ)介質(zhì),而磁盤則作為備份介質(zhì),為了提高效率,RAMCloud使用日志結(jié)構(gòu)存儲(chǔ),利用集群的并發(fā)能力和高速帶寬的Infiniband實(shí)現(xiàn)快速恢復(fù),保證了數(shù)據(jù)的可用性和完整性,但Spark與RAMCloud并不兼容,因?yàn)镽AMCloud的高內(nèi)存占用率會(huì)影響Spark的計(jì)算性能.

    因此,為了解決檢查點(diǎn)設(shè)置的問(wèn)題,其中的重點(diǎn)就是選擇哪些有價(jià)值的RDD作為檢查點(diǎn)存儲(chǔ)在磁盤中,以及在數(shù)據(jù)丟失后應(yīng)選擇什么樣的節(jié)點(diǎn)進(jìn)行恢復(fù).在實(shí)時(shí)系統(tǒng)或數(shù)據(jù)庫(kù)系統(tǒng)中,對(duì)周期設(shè)置檢查點(diǎn)而言,所有數(shù)據(jù)的重要性相同,僅僅根據(jù)周期設(shè)置時(shí)的時(shí)間選擇最新生成的數(shù)據(jù)設(shè)置為檢查點(diǎn).與此不同的是,由于Spark基本存儲(chǔ)單元為RDD,不同RDD所具有的屬性如操作復(fù)雜度、血統(tǒng)長(zhǎng)度、計(jì)算代價(jià)和RDD數(shù)據(jù)量大小各不相同,對(duì)于作業(yè)恢復(fù)時(shí)所產(chǎn)生的恢復(fù)效果不同,因此在作業(yè)執(zhí)行過(guò)程中,異步存儲(chǔ)對(duì)恢復(fù)更有價(jià)值的RDD在磁盤中,在設(shè)置檢查點(diǎn)的同時(shí)盡量不影響作業(yè)執(zhí)行效率.通過(guò)對(duì)Spark作業(yè)執(zhí)行機(jī)制和執(zhí)行時(shí)間進(jìn)行理論分析,在恢復(fù)時(shí)根據(jù)數(shù)據(jù)本地性、節(jié)點(diǎn)空閑情況和節(jié)點(diǎn)能力選擇節(jié)點(diǎn),并執(zhí)行恢復(fù)任務(wù),從而縮短恢復(fù)時(shí)間.

    2 問(wèn)題的建模和分析

    本節(jié)通過(guò)分析Spark作業(yè)執(zhí)行機(jī)制,定義了RDD和作業(yè)的執(zhí)行開銷,提出了作業(yè)執(zhí)行模型、檢查點(diǎn)模型和關(guān)鍵度模型,并且在理論基礎(chǔ)上對(duì)目標(biāo)問(wèn)題進(jìn)行了定義.

    2.1 Spark作業(yè)執(zhí)行機(jī)制

    在Spark應(yīng)用中,整個(gè)執(zhí)行流程在邏輯上會(huì)形成有向無(wú)環(huán)圖(DAG).Action算子觸發(fā)之后,根據(jù)RDD的血統(tǒng),將所有累積的算子形成一個(gè)有向無(wú)環(huán)圖,然后由調(diào)度器調(diào)度該圖上的任務(wù)進(jìn)行運(yùn)算.Spark根據(jù)RDD之間不同的依賴關(guān)系切分形成不同的階段(stage),一個(gè)階段包含一系列函數(shù)執(zhí)行流水線.

    Spark作業(yè)DAG的典型示例如圖1所示,A,B,C,D,E,F(xiàn)和G分別代表不同的RDD,RDD內(nèi)的方框代表分區(qū),虛線框?yàn)殡A段.數(shù)據(jù)從HDFS輸入Spark,形成RDDA和RDDC,RDDC上執(zhí)行map操作,轉(zhuǎn)換為RDDD,RDDD和RDDE執(zhí)行union操作,轉(zhuǎn)換為F,而在B和F通過(guò)join連接轉(zhuǎn)化為G的過(guò)程中會(huì)執(zhí)行Shuffle,最后RDDG輸出并保存到HDFS中.

    Fig. 1 Directed acyclic graph of Spark job圖1 Spark作業(yè)的有向無(wú)環(huán)圖

    Spark在未設(shè)置檢查點(diǎn)算法時(shí),若數(shù)據(jù)失效,則需要利用血統(tǒng)重新計(jì)算來(lái)實(shí)現(xiàn)恢復(fù).若所有需要的數(shù)據(jù)都丟失,則RDD必須重新計(jì)算生成,使任務(wù)的完成時(shí)間延長(zhǎng),還增加了額外的計(jì)算開銷.若能為有價(jià)值的RDD設(shè)置檢查點(diǎn),則能有效縮減任務(wù)恢復(fù)時(shí)間.

    2.2 作業(yè)執(zhí)行模型

    定義1. RDD計(jì)算代價(jià).Spark任務(wù)中,分區(qū)由父節(jié)點(diǎn)為輸入數(shù)據(jù)計(jì)算生成.設(shè)Parentsi jk為分區(qū)Pi jk的父節(jié)點(diǎn)集合.分區(qū)Pi jk的計(jì)算代價(jià)為數(shù)據(jù)讀取代價(jià)與數(shù)據(jù)處理代價(jià)之和,即:

    TPi jk=read(Parentsi jk)+proc(Parentsi jk).

    (1)

    RDD的所有分區(qū)由集群工作節(jié)點(diǎn)并行計(jì)算生成,因此其計(jì)算代價(jià)為所有分區(qū)計(jì)算代價(jià)的最大值,即:

    TRDDi j=max(TPi j1,TPi j2,…,TPi jk).

    (2)

    定義2. 作業(yè)執(zhí)行時(shí)間.如圖1所示,Spark以寬依賴為分界點(diǎn),將作業(yè)劃分為多個(gè)階段執(zhí)行,因此階段分為窄依賴和寬依賴2類.

    由于Spark任務(wù)以分區(qū)為最小粒度單位,因此在任務(wù)分配時(shí),分區(qū)計(jì)算時(shí)間與節(jié)點(diǎn)計(jì)算能力有關(guān).設(shè)cpw為工作節(jié)點(diǎn)計(jì)算能力,單位是tuple/s;SPi jk為分區(qū)Pi jk的元組個(gè)數(shù).

    由于窄依賴的父分區(qū)為上一個(gè)RDD對(duì)應(yīng)位置的一個(gè)分區(qū),則窄依賴分區(qū)計(jì)算時(shí)間:

    (3)

    對(duì)于窄依賴階段,每個(gè)階段包括多條流水線(每條流水線包括多個(gè)RDD的不同分區(qū)).記窄依賴Stage共包含m個(gè)RDD,RDD相同位置的分區(qū)組成流水線,所有RDD劃分為x條流水線,單條流水線的分區(qū)集合為pipei x={Pi 1x,Pi 2x,…,Pi jx},那么單條流水線的執(zhí)行時(shí)間可表示為

    (4)

    對(duì)于stagei,記其流水線集合為Pipesi={pipei 1,pipei 2,…,pipei x},那么該階段的執(zhí)行時(shí)間應(yīng)為各流水線執(zhí)行時(shí)間最大值,即:

    NTstagei=max(Tpipei 1,Tpipei 2,…,Tpipei x)=

    寬依賴的父分區(qū)集合為上一個(gè)RDD的所有分區(qū),即整個(gè)RDD,且寬依賴RDD的分區(qū)計(jì)算時(shí)間大于窄依賴RDD的分區(qū)計(jì)算時(shí)間,則寬依賴分區(qū)計(jì)算時(shí)間:

    (6)

    由于寬依賴階段中僅包含1個(gè)RDD,則寬依賴階段的計(jì)算時(shí)間為

    WTstagei=max{WTPi j1,WTPi j2,…,WTPi jk}.

    (7)

    一個(gè)作業(yè)由若干個(gè)階段構(gòu)成,由于Spark以寬依賴為界劃分階段,而作業(yè)的最后一個(gè)階段必為寬依賴(所有執(zhí)行均為寬依賴操作),記作業(yè)中寬、窄依賴的個(gè)數(shù)分別為m和n,那么作業(yè)執(zhí)行時(shí)間可表示為

    (8)

    2.3 檢查點(diǎn)模型

    在節(jié)點(diǎn)故障時(shí),會(huì)使存儲(chǔ)在該節(jié)點(diǎn)內(nèi)存上的多個(gè)RDD部分甚至所有分區(qū)不可用,這將導(dǎo)致作業(yè)無(wú)法繼續(xù)正常執(zhí)行,在未設(shè)置檢查點(diǎn)的情況下,作業(yè)需要回溯,直到找到可用的RDD.極端情況下,甚至需要重新調(diào)度,之前所計(jì)算的所有工作和耗費(fèi)的系統(tǒng)資源都會(huì)浪費(fèi).因此,對(duì)于作業(yè)而言,有必要設(shè)置檢查點(diǎn),從而降低節(jié)點(diǎn)宕機(jī)后產(chǎn)生的恢復(fù)開銷.

    定義3. 作業(yè)恢復(fù)代價(jià).作業(yè)執(zhí)行到任意RDD,若集群中某個(gè)工作節(jié)點(diǎn)發(fā)生故障,作業(yè)將丟失該工作節(jié)點(diǎn)計(jì)算的所有中間結(jié)果.記故障的工作節(jié)點(diǎn)編號(hào)為k(工作節(jié)點(diǎn)編號(hào)對(duì)應(yīng)流水線序號(hào)和寬依賴RDD中的分區(qū)編號(hào)),那么對(duì)于所有窄依賴階段,每個(gè)階段丟失一條流水線的數(shù)據(jù),則在不考慮恢復(fù)調(diào)度開銷時(shí),窄依賴階段的恢復(fù)代價(jià)為丟失流水線的重新計(jì)算代價(jià),即:

    (9)

    而對(duì)于寬依賴階段而言,RDD的一個(gè)分區(qū)丟失,因此在不考慮恢復(fù)調(diào)度開銷時(shí),寬依賴階段的恢復(fù)代價(jià)為該分區(qū)的重新計(jì)算代價(jià),即:

    WRi=WTPi jk.

    (10)

    由于工作節(jié)點(diǎn)故障為隨機(jī)事件,發(fā)生故障時(shí)當(dāng)前計(jì)算的RDD可能位于任何階段內(nèi),因此,若故障發(fā)生在窄依賴階段,則作業(yè)恢復(fù)代價(jià)還應(yīng)考慮丟失分區(qū)在流水線的前續(xù)節(jié)點(diǎn)的重新計(jì)算代價(jià).而若故障發(fā)生在寬依賴階段,由于該階段僅包含一個(gè)RDD,無(wú)前續(xù)節(jié)點(diǎn)恢復(fù)代價(jià),僅需要恢復(fù)該RDD.

    設(shè)當(dāng)前作業(yè)已執(zhí)行的階段中共有x個(gè)窄依賴、y個(gè)寬依賴,正在執(zhí)行的階段共有m個(gè)RDD,當(dāng)前正在計(jì)算RDDi h,在不考慮恢復(fù)調(diào)度開銷時(shí)作業(yè)恢復(fù)代價(jià)可表示為

    (11)

    即作業(yè)恢復(fù)代價(jià)等于前續(xù)寬、窄依賴階段的恢復(fù)代價(jià)之和,再加上丟失分區(qū)在當(dāng)前階段中父分區(qū)的計(jì)算代價(jià)之和.

    定理1. 設(shè)Jobi={RDDi 1,RDDi 2,…,RDDi n},RDDi(j-1)∈Jobi且RDDi j∈Jobi.若RDDi j具有更長(zhǎng)的血統(tǒng),并且RDDi(j-1)是RDDi j唯一的parentRDD.那么,在執(zhí)行Jobi的計(jì)算節(jié)點(diǎn)都失效時(shí),恢復(fù)RDDi j的開銷較高,且為TRDDi j.

    證明. 若RDDi j具有更長(zhǎng)的血統(tǒng),RDDi(j-1)是RDDi j的parentRDD.設(shè)RDDi(j-1)的血統(tǒng)長(zhǎng)度為k,則RDDi j的血統(tǒng)長(zhǎng)度為k+1.考慮2種情況:

    1) 若Jobi設(shè)置了檢查點(diǎn),則只需要從檢查點(diǎn)處開始計(jì)算恢復(fù).若恢復(fù)所需要的最新檢查點(diǎn)為RDDi p,恢復(fù)調(diào)度開銷為常量αi的情況下,則恢復(fù)開銷分別為

    2) 若Jobi未設(shè)置檢查點(diǎn),在計(jì)算節(jié)點(diǎn)失效時(shí),則所有RDD都需要重新計(jì)算,重新部署調(diào)度開銷為常量βi,則恢復(fù)開銷分別為

    無(wú)論是情況1或情況2,明顯可看出,RRDDi j>RRDDi(j-1)且RRDDi j-RRDDi(j-1)=TRDDi j.

    證畢.

    定理2. 設(shè)Jobi={RDDi 1,RDDi 2,…,RDDi n},RDDi(j-1)∈Jobi,且RDDi j∈Jobi.RDDi(j-1)為RDDi j的唯一parentRDD.若節(jié)點(diǎn)失效時(shí),僅丟失了RDDi j的第l個(gè)分區(qū)Pi jl,那么在RDDi j的操作為窄依賴或?qū)捯蕾嚂r(shí),其恢復(fù)Pi jl的開銷不同,且寬依賴時(shí)恢復(fù)開銷為max(RPi j1,RPi j2,…,RPi jk).

    證明. 由于RDDi j的parentRDD為RDDi(j-1),則操作為窄依賴時(shí)可知丟失第l個(gè)分片Pi jl,只需通過(guò)RDDi(j-1)的對(duì)應(yīng)父分片Pi j(l-1)通過(guò)流水線計(jì)算獲得,故恢復(fù)Pi j(l-1)開銷為

    RRDDi j(narrow)=RPTi jl=
    αi+read(PTi(j-1)l)+proc(PTi jl).

    操作為寬依賴時(shí),丟失第l個(gè)分片Pi jl,需通過(guò)RDDi(j-1)所有的分片計(jì)算得到Pi jl,則恢復(fù)PTi j(l-1)的開銷為

    RRDDi j(wide)=RRDDi j=max(RPi j1,RPi j2,…,RPi jk).

    因此,RRDDi j(wide)≥RRDDi j(narrow).

    證畢.

    定理3. 設(shè)Jobi={RDDi 1,RDDi 2,…,RDDi n},RDDi(j-1),RDDi j,RDDi(j+1)∈Jobi.RDDi(j-1)是RDDi(j-1)和RDDi j的唯一parentRDD,RDDi(j-1)已備份為檢查點(diǎn)并存儲(chǔ)在正常運(yùn)行節(jié)點(diǎn),且計(jì)算代價(jià)procRDDi(j+1)≥procRDDi j.當(dāng)計(jì)算節(jié)點(diǎn)失效時(shí),RDDi j和RDDi(j+1)丟失,則從RDDi(j-1)檢查點(diǎn)進(jìn)行恢復(fù)的代價(jià)RRDDi(j+1)≥RRDDi j.

    證明. 由于procRDDi(j+1)≥procRDDi j,則對(duì)于RDDi(j+1)中任意第l個(gè)分片的計(jì)算代價(jià)procPTi(j+1)l≥procPTi jl.

    當(dāng)執(zhí)行恢復(fù)計(jì)算時(shí),RDDi(j-1)已備份為檢查點(diǎn),則恢復(fù)RDDi j的開銷為RRDDi j=max(RPi j1,RPi j2,…,RPi jk).對(duì)第l個(gè)分片而言,恢復(fù)該分片的開銷為

    RPi jl=TPi jl=αi+read(Pi(j-1)l)+proc(Pi jl).

    同樣,恢復(fù)RDDi(j+1)的開銷為RRDDi(j+1)=max(RPi(j+1)1,RPi(j+1)2,…,RPi(j+1)k).對(duì)第l個(gè)分片而言,恢復(fù)該分片的開銷為

    RPi(j+1)l=TPi(j+1)l=
    αi+read(Pi(j-1)l)+proc(Pi(j+1)l).

    由于procRDDi(j+1)≥procRDDi j,可知RPi(j+1)l≥RPi jl,同時(shí)RRDDi(j+1)≥RRDDi j.

    證畢.

    2.4 RDD關(guān)鍵度模型

    根據(jù)2.3節(jié)的分析,可以判斷出RDD與恢復(fù)開銷相關(guān)的因素有3個(gè):

    1) RDD的血統(tǒng)長(zhǎng)度.在作業(yè)中具有更長(zhǎng)血統(tǒng)的RDD恢復(fù)開銷更大,證明詳見(jiàn)定理1.因此當(dāng)設(shè)置檢查點(diǎn)時(shí),應(yīng)優(yōu)先選擇長(zhǎng)血統(tǒng)的RDD,從而降低對(duì)恢復(fù)開銷的影響.

    2) RDD類型(指寬依賴或窄依賴).在寬依賴情況下,丟失1個(gè)子RDD分區(qū),需重算的每個(gè)父RDD的每個(gè)分區(qū)會(huì)產(chǎn)生冗余計(jì)算開銷,使寬依賴在恢復(fù)時(shí)開銷更大,證明詳見(jiàn)定理2.因此,應(yīng)優(yōu)先選擇寬依賴的RDD作為檢查點(diǎn),從而降低恢復(fù)開銷.

    3) RDD計(jì)算代價(jià).計(jì)算代價(jià)越高的RDD恢復(fù)時(shí)開銷越大,應(yīng)優(yōu)先備份計(jì)算代價(jià)高的RDD,證明詳見(jiàn)定理3.另外,RDD大小對(duì)數(shù)據(jù)備份時(shí)長(zhǎng)和數(shù)據(jù)恢復(fù)讀取時(shí)長(zhǎng)都會(huì)產(chǎn)生影響,從而也作為考慮的關(guān)鍵因素.

    定義4. RDD關(guān)鍵度.表示RDD對(duì)任務(wù)恢復(fù)效率的重要程度.在傳統(tǒng)的檢查點(diǎn)策略中,并未引進(jìn)權(quán)重的概念,被選擇作為檢查點(diǎn)的數(shù)據(jù)在本質(zhì)上與其他數(shù)據(jù)沒(méi)有區(qū)別,而在Spark中則不同,不同的RDD對(duì)恢復(fù)的重要程度不同.在RDD丟失時(shí),需要重新計(jì)算,而不同的RDD恢復(fù)需要的計(jì)算開銷不同.

    綜合考慮相關(guān)因素,記LDRDDi j表示為L(zhǎng)DRDDi j血統(tǒng)長(zhǎng)度,TypeRDDi j為類型,CostRDDi j為計(jì)算代價(jià),SizeRDDi j為數(shù)據(jù)量大小.則RDD的關(guān)鍵度表示為

    (12)

    式(12)表明,血統(tǒng)長(zhǎng)度、RDD類型、計(jì)算代價(jià)與RDD關(guān)鍵度成正比關(guān)系,而容量大小則成反比關(guān)系.關(guān)鍵度越大,恢復(fù)時(shí)對(duì)恢復(fù)效率的影響越大,因此作為檢查點(diǎn)備份的必要性越大.

    定義5. 失效恢復(fù)比.用于表示恢復(fù)任務(wù)分配與節(jié)點(diǎn)計(jì)算能力的適應(yīng)程度.記恢復(fù)計(jì)算節(jié)點(diǎn)集合Workers={w1,w2,…,wn},其計(jì)算能力C={c1,c2,…,cn},RT為恢復(fù)任務(wù)總量,則執(zhí)行恢復(fù)任務(wù)的時(shí)間開銷均值可定義為

    (13)

    對(duì)于任意計(jì)算節(jié)點(diǎn)wi,分配給其恢復(fù)任務(wù)RTi時(shí),其執(zhí)行時(shí)間、均值的方差和失效恢復(fù)比分別為

    (14)

    Dwi=(Twi-E)2,

    (15)

    (16)

    其中,wi∈Workers.

    基于定義,從作業(yè)分配的角度來(lái)看,作業(yè)執(zhí)行時(shí)間也可表示為

    Tjob=max(Tw1,Tw2,…,Twn).

    (17)

    由于節(jié)點(diǎn)的失效恢復(fù)比與方差成反比,因此比值越大,方差越小,表示節(jié)點(diǎn)恢復(fù)作業(yè)完成時(shí)間越趨近均值,因此當(dāng)所有工作節(jié)點(diǎn)的失效恢復(fù)比取最大值時(shí)恢復(fù)任務(wù)的執(zhí)行時(shí)間最短.

    2.5 檢查點(diǎn)管理策略問(wèn)題定義

    2.1節(jié)至2.4節(jié)已經(jīng)對(duì)作業(yè)和檢查點(diǎn)機(jī)制作了比較詳細(xì)的闡述,基于這些定義,對(duì)我們的檢查點(diǎn)算法進(jìn)行形式化表示.

    優(yōu)化目標(biāo):

    minRjob.

    (18)

    s.t.

    (19)

    其中,j∈jobs,wi∈Workers,r∈R.

    優(yōu)化目標(biāo):

    (20)

    (21)

    s.t.

    (22)

    其中,j∈jobs,RDDj k∈jobj,wi∈Workers,r∈R.

    目標(biāo)是最大化RDD關(guān)鍵度和失效恢復(fù)比,約束條件同上.因此,在設(shè)置檢查點(diǎn)時(shí),應(yīng)選擇已生成的RDD中對(duì)恢復(fù)效率影響最大,即關(guān)鍵度最大的RDD作為檢查點(diǎn)進(jìn)行設(shè)置.在節(jié)點(diǎn)失效進(jìn)行恢復(fù)時(shí),應(yīng)選擇節(jié)點(diǎn)能力更強(qiáng)的節(jié)點(diǎn)執(zhí)行作業(yè)恢復(fù).

    3 檢查點(diǎn)管理策略

    本節(jié)提出基于關(guān)鍵度的檢查點(diǎn)管理(criticality based checkpoint management,CCM)策略,其中包括檢查點(diǎn)設(shè)置算法、失效恢復(fù)算法和檢查點(diǎn)清理算法,具體流程圖如圖2所示.

    Fig. 2 Flow chart of checkpoint strategy圖2 檢查點(diǎn)策略流程圖

    作業(yè)在設(shè)置檢查點(diǎn)和恢復(fù)算法的粒度可以分為2個(gè)級(jí)別:1)RDD級(jí)別;2)分區(qū)級(jí)別.在設(shè)置檢查點(diǎn)時(shí),檢查點(diǎn)選擇可以在RDD級(jí)別、分區(qū)級(jí)別以權(quán)重度、固定時(shí)間間隔等多種為算法以異步檢查點(diǎn)的方式進(jìn)行設(shè)置.在算法恢復(fù)時(shí),宕機(jī)節(jié)點(diǎn)上的任務(wù)需要進(jìn)行恢復(fù),以最新的檢查點(diǎn)序列為基礎(chǔ),由Spark調(diào)度器重新調(diào)度執(zhí)行.根據(jù)集群資源情況,分配需要的資源.

    RDD級(jí)別以作業(yè)的RDD為粒度,根據(jù)RDD的重要度選擇更優(yōu)的檢查點(diǎn),問(wèn)題在于粒度較大,當(dāng)某個(gè)節(jié)點(diǎn)宕機(jī)丟失某些RDD的分區(qū),需要重新計(jì)算所有的RDD,從磁盤中讀取最新檢查點(diǎn)信息的所有內(nèi)容,浪費(fèi)系統(tǒng)資源.分區(qū)級(jí)別以RDD的分區(qū)為粒度,根據(jù)分區(qū)的關(guān)鍵度,為每個(gè)分區(qū)選擇異步的檢查點(diǎn),在丟失某些分區(qū)時(shí)不需要恢復(fù)整個(gè)RDD,只需恢復(fù)丟失的分區(qū),無(wú)論是備份檢查點(diǎn)時(shí)的磁盤開銷,還是恢復(fù)時(shí)的讀取開銷和執(zhí)行開銷,都比較少.但問(wèn)題在于,以分區(qū)為粒度和異步備份檢查點(diǎn)時(shí)、若有多個(gè)任務(wù)同時(shí)在Spark集群中執(zhí)行,檢查點(diǎn)設(shè)置開銷和管理難度明顯增大.因此,采用以RDD為設(shè)置檢查點(diǎn)的基本單位,分區(qū)為恢復(fù)檢查點(diǎn)的基本單位.

    3.1 基礎(chǔ)數(shù)據(jù)構(gòu)建

    基于關(guān)鍵度的檢查點(diǎn)算法需要構(gòu)建的初始化參數(shù)與基礎(chǔ)數(shù)據(jù)如下:

    1) 生成RDD結(jié)構(gòu)樹treeRDDs

    通過(guò)在Spark源碼中插入監(jiān)聽代碼,遍歷作業(yè)的DAG圖,獲取DAG圖中點(diǎn)和邊的信息,以及輸入RDDId、操作類型、輸出RDDId,并通過(guò)DAG生成RDD結(jié)構(gòu)樹treeRDDs.

    2) 獲取RDD深度

    通過(guò)分析DAG圖即可獲得,實(shí)際運(yùn)行時(shí)血統(tǒng)長(zhǎng)度為迭代次數(shù)與血統(tǒng)長(zhǎng)度相乘.

    3) 獲取RDD類型值

    寬依賴操作的RDD具有作為檢查點(diǎn)的意義,因此遍歷RDD結(jié)構(gòu)樹,將操作為寬依賴的RDD寫入列表,并根據(jù)該RDD分片個(gè)數(shù)獲得類型值.

    4) RDD計(jì)算代價(jià)

    在實(shí)際的Spark環(huán)境中,作業(yè)的DAG圖生成后,管理節(jié)點(diǎn)監(jiān)控所有RDD的狀態(tài)變化,通過(guò)記錄各狀態(tài)變化的時(shí)間點(diǎn)即可得出起始時(shí)間和完成時(shí)間.

    5) RDD大小

    在作業(yè)執(zhí)行過(guò)程中,每計(jì)算完成一個(gè)RDD,即可根據(jù)其多個(gè)分區(qū)大小求和,進(jìn)而獲得該RDD的大小.

    6) 計(jì)算關(guān)鍵度

    通過(guò)RDD關(guān)鍵度公式,為RDD計(jì)算并保存.

    3.2 檢查點(diǎn)設(shè)置算法

    如算法1所示,根據(jù)RDD的關(guān)鍵度,在作業(yè)執(zhí)行時(shí)間內(nèi)選擇RDD作為檢查點(diǎn)進(jìn)行設(shè)置.檢查點(diǎn)設(shè)置的時(shí)間從源RDD之后第1個(gè)生成的RDD開始作為檢查點(diǎn)開始,到終點(diǎn)RDD之前最后一個(gè)生成的RDD結(jié)束為結(jié)束.在作業(yè)執(zhí)行時(shí),源RDD和終點(diǎn)RDD不需要進(jìn)行檢查點(diǎn)設(shè)置.由于源RDD存儲(chǔ)在磁盤中,終點(diǎn)RDD為任務(wù)結(jié)束的RDD,會(huì)根據(jù)需要在結(jié)束后寫入磁盤,因此不需要考慮.在作業(yè)執(zhí)行過(guò)程中進(jìn)行檢查點(diǎn)設(shè)置時(shí)對(duì)比最新生成的多個(gè)RDD,并選擇當(dāng)前關(guān)鍵度最大的RDD.選擇好需要備份的RDD,執(zhí)行檢查點(diǎn)設(shè)置操作,將該RDD備份.計(jì)算任務(wù)首先將檢查點(diǎn)臨時(shí)緩存在局部?jī)?nèi)存中,然后由另一個(gè)獨(dú)立于計(jì)算任務(wù)的并行任務(wù)負(fù)責(zé)將緩存在內(nèi)存中的檢查點(diǎn)數(shù)據(jù)文件拷貝到HDFS.

    算法1. 檢查點(diǎn)設(shè)置算法.

    輸入:結(jié)構(gòu)樹treeRDDs;

    輸出:檢查點(diǎn)列表checkpointlist.

    初始化:checkpointlist←newArray;

    checkpoint←newHash;

    cutlist←newList;

    i←0;

    j←1;

    maxCR←0.

    ①checkpointlist←Null;

    ②checkpoint(treeRDDs[i]);

    ③checkpointlist.add(treeRDDs[i]);

    /*添加生成的第1個(gè)RDD作為檢查點(diǎn)*/

    ④ while(Cfinished=1 &&Tfinished=0)

    ⑤ fori=0 tonewgenerateRDD.lengthdo

    ⑥candidateslist←getnewRDD;

    ⑦candidates[i].cost←getRDDCost;

    ⑧calculate(candidates[i].CR);

    ⑨ if (candidates[i].CR>maxCR) then

    ⑩nextcheckpoint[j]←candidates[i];

    3.3 檢查點(diǎn)恢復(fù)算法

    管理節(jié)點(diǎn)在每一個(gè)固定的時(shí)間間隔要求工作節(jié)點(diǎn)來(lái)發(fā)送它們的當(dāng)前狀態(tài).工作節(jié)點(diǎn)定期發(fā)送心跳給管理節(jié)點(diǎn),如果在一個(gè)時(shí)間限制沒(méi)有收到工作節(jié)點(diǎn)的心跳,標(biāo)記“失敗”.在這種情況下,管理節(jié)點(diǎn)指示工作節(jié)點(diǎn)利用最近的檢查點(diǎn)進(jìn)行重新啟動(dòng)恢復(fù).如算法2所示,宕機(jī)后,由Spark執(zhí)行恢復(fù)操作時(shí),從最后設(shè)置的檢查點(diǎn)RDD處進(jìn)行恢復(fù).恢復(fù)時(shí),優(yōu)先選擇離執(zhí)行恢復(fù)的工作節(jié)點(diǎn)最近的檢查點(diǎn)副本,從而降低網(wǎng)絡(luò)開銷和恢復(fù)時(shí)延.將最新的檢查點(diǎn)讀入內(nèi)存,并且可以根據(jù)需要將檢查點(diǎn)列表中的檢查點(diǎn)讀入,從而降低恢復(fù)和執(zhí)行開銷.當(dāng)某個(gè)RDD需要恢復(fù)但未設(shè)置檢查點(diǎn)時(shí),重新執(zhí)行血統(tǒng),通過(guò)其父節(jié)點(diǎn)恢復(fù);若已將該RDD設(shè)置檢查點(diǎn),可以將該RDD讀入內(nèi)存.若有寬依賴,或丟失了RDD的所有分區(qū),則需讀取所需的RDD的所有分區(qū)到內(nèi)存;若丟失了RDD的部分分區(qū),且恢復(fù)血統(tǒng)中無(wú)寬依賴,則只需讀取丟失分區(qū)所需檢查點(diǎn)到內(nèi)存進(jìn)行計(jì)算.

    算法2. 檢查點(diǎn)恢復(fù)算法.

    輸入:當(dāng)前結(jié)構(gòu)樹curtreeRDDs、檢查點(diǎn)列表checkpointlist、需要恢復(fù)的RDDrecoveryRDDs.

    初始化:parentsRDD←newHash;

    checkpointRDD←newHash;

    i,j,k←0;

    recoveryworker←assign.freeworker.

    ① fori=0 torecoveryRDDs.length-1 do

    ②parentsRDD←getLineage(

    recoveryRDDs[i],curtreeRDDs);

    /*獲取需要恢復(fù)RDD的父RDD*/

    ③ forj=0 toparentsRDD.length-1 do

    ④checkpointRDD←get(checkpointlist,

    parentsRDD);

    ⑤ fork=0 torecoveryRDDs.partitionnum-1 do

    ⑥ if (wideDependency(recoveryRDDs[i])∨lost(recoveryRDDs[i])

    ⑦read(parentsRDD[j]);

    /*操作為寬依賴或丟失所有分區(qū)*/

    ⑧ end if

    ⑩read(parentsRDD[j].partition[k]);

    /*操作為窄依賴且丟失部分分區(qū)*/

    /*根據(jù)血統(tǒng)進(jìn)行恢復(fù)計(jì)算*/

    3.4 檢查點(diǎn)清理算法

    隨著任務(wù)執(zhí)行時(shí)間的增長(zhǎng),保存的檢查點(diǎn)恢復(fù)信息越來(lái)越多,其中有些恢復(fù)信息成為過(guò)時(shí)無(wú)用的恢復(fù)信息.檢查點(diǎn)清理就是刪除這些過(guò)時(shí)無(wú)用的恢復(fù)信息.檢查點(diǎn)清理策略對(duì)恢復(fù)協(xié)議性能有直接影響.不同的恢復(fù)策略,不同的應(yīng)用所需要存儲(chǔ)的恢復(fù)信息量都會(huì)有不同,執(zhí)行檢查點(diǎn)清理會(huì)伴隨有開銷.因此有效的組織存儲(chǔ)恢復(fù)信息和執(zhí)行檢查點(diǎn)清理都會(huì)對(duì)系統(tǒng)性能產(chǎn)生影響,提高集群資源利用率.

    對(duì)Spark檢查點(diǎn)機(jī)制而言,檢查點(diǎn)存儲(chǔ)在持久化存儲(chǔ)磁盤或固態(tài)硬盤中.就磁盤空間而言,空間較大,并不是主要的瓶頸.因此,最簡(jiǎn)單的做法就是在任務(wù)執(zhí)行完畢后,將該任務(wù)所設(shè)置的所有檢查點(diǎn)刪除.但每次檢查點(diǎn)備份時(shí)為3個(gè)副本,因此,當(dāng)系統(tǒng)中并行執(zhí)行多個(gè)任務(wù)時(shí),若磁盤空間有限,那么在備份過(guò)程中需要執(zhí)行清理算法,可以根據(jù)檢查點(diǎn)的關(guān)鍵度進(jìn)行清理.

    檢查點(diǎn)清理算法,在不影響其他作業(yè)的情況下并行執(zhí)行.如算法3所示,清理的2個(gè)條件為:

    1) 當(dāng)任務(wù)的磁盤備份空間受限時(shí),在添加新檢查點(diǎn)時(shí),對(duì)已有檢查點(diǎn)進(jìn)行清理,以滿足新檢查點(diǎn)的空間需求;

    2) 系統(tǒng)設(shè)置檢查點(diǎn)清理周期,到達(dá)該時(shí)間周期時(shí)進(jìn)行清理.

    清理算法的2個(gè)步驟:

    1) 新添加一個(gè)檢查點(diǎn),磁盤空間不足,將已有的檢查點(diǎn)關(guān)鍵度進(jìn)行過(guò)濾,將權(quán)重小于新檢查點(diǎn)的對(duì)象放入候選列表.

    2) 將候選列表按關(guān)鍵度從小到大的順序排列.搜索候選列表,若存在目標(biāo),其容量滿足新檢查點(diǎn)的要求則進(jìn)行清理;否則,當(dāng)用戶設(shè)定的清理時(shí)間到達(dá)時(shí),才能將所有候列表中的檢查點(diǎn)進(jìn)行清理.

    算法3. 檢查點(diǎn)清理算法.

    輸入:檢查點(diǎn)列表checkpointlist、當(dāng)前需添加的檢查點(diǎn)RDD[i]、存儲(chǔ)空間空閑容量freecapacity.

    初始化:candidates←newHash;

    v←RDD[i].CR;

    s←RDD[i].size.

    ① forj=0 tocheckpointlist.length-1 do

    ② ifv>checkpointlist[j].CRthen

    ③candidates.add(checkpointlist[j]);

    ④ end if

    ⑤ end for

    ⑥cancadidates.orderbyCR();

    /*對(duì)關(guān)鍵度進(jìn)行排序*/

    ⑦ forj=0 tocancadidates.length-1 do

    ⑧ if (freecapacity≤mincapacity∧

    cancadidates[j].size>s)

    ⑨clean(candidates[j]);

    ⑩checkpoint(RDDs[i]);

    /*選擇空間滿足大小的檢查點(diǎn)替換*/

    /*到達(dá)系統(tǒng)清理時(shí)間*/

    /*清理所有檢查點(diǎn)*/

    4 實(shí)驗(yàn)與評(píng)價(jià)

    本節(jié)將通過(guò)實(shí)驗(yàn)進(jìn)行比較和評(píng)價(jià),驗(yàn)證檢查點(diǎn)自動(dòng)選擇算法、恢復(fù)算法和檢查點(diǎn)清理算法的有效性.

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

    實(shí)驗(yàn)環(huán)境用1臺(tái)服務(wù)器和8個(gè)工作節(jié)點(diǎn)建立計(jì)算集群,服務(wù)器作為Spark的Master和Hadoop的NameNode.為體現(xiàn)工作節(jié)點(diǎn)的計(jì)算能力不同,8個(gè)工作節(jié)點(diǎn)由1個(gè)高效節(jié)點(diǎn)、6個(gè)普通節(jié)點(diǎn)和1個(gè)慢節(jié)點(diǎn)組成,其中普通節(jié)點(diǎn)的配置如表1所示,高效節(jié)點(diǎn)配備4核CPU,16 GB內(nèi)存和4個(gè)千兆網(wǎng)卡,而慢任務(wù)節(jié)點(diǎn)僅有單核CPU,1 GB內(nèi)存和1個(gè)百兆網(wǎng)卡.

    Table1 Configuration Parameters of Worker Node表1 Worker節(jié)點(diǎn)配置參數(shù)

    4.2 算法綜合評(píng)估測(cè)試

    對(duì)于本文提出CCM策略進(jìn)行測(cè)試,實(shí)驗(yàn)數(shù)據(jù)首先選取WordCount,TeraSort,K-Means,PageRank 4種算法作為作業(yè)進(jìn)行分析.其中WordCount作業(yè)量為4.9 GB,TeraSort作業(yè)量為4.5 GB,K-Means輸入數(shù)據(jù)總樣本點(diǎn)為4 000 000,維度為20,點(diǎn)群個(gè)數(shù)為5,迭代1次.PageRank頁(yè)面?zhèn)€數(shù)為3 000 000,迭代3次.表2為不同算法在集群失效節(jié)點(diǎn)個(gè)數(shù)為1時(shí)執(zhí)行,且執(zhí)行到第100秒時(shí)關(guān)閉1個(gè)普通節(jié)點(diǎn)node3,對(duì)比3種策略對(duì)恢復(fù)時(shí)間(execution time, ET)以及與未優(yōu)化策略相比的加速時(shí)間(accelerate time, AT)和加速比(accelerate rate, AR)的影響.這3種策略分別為未優(yōu)化策略(Without opt)、Tachyon檢查點(diǎn)策略(Tachyon checkpoint)和CCM策略.

    Table 2 Recovery Efficiency with Different Strategies inSingle Node Failure

    在所有的作業(yè)中,CCM策略的作業(yè)完成時(shí)間都優(yōu)于傳統(tǒng)Spark,在設(shè)置檢查點(diǎn)的情況下,宕機(jī)恢復(fù)后,4類作業(yè)的總執(zhí)行時(shí)間要小于未設(shè)置檢查點(diǎn)的情況,從而證明了算法在多種類型作業(yè)下都具有良好的優(yōu)化效果.通過(guò)觀察數(shù)據(jù)發(fā)現(xiàn),不同作業(yè)對(duì)恢復(fù)效率的提高程度各不相同,這是因?yàn)椴煌愋妥鳂I(yè)寬依賴操作個(gè)數(shù)和數(shù)據(jù)大小各不相同,因此在作業(yè)類型不同的情況下算法的優(yōu)化效果無(wú)明顯規(guī)律.

    Fig. 3 Memory usage of different jobs on node3圖3 普通節(jié)點(diǎn)node3上不同作業(yè)的內(nèi)存利用率

    Fig. 4 Disk I/O rate of different jobs on node3圖4 普通節(jié)點(diǎn)node3上不同作業(yè)的磁盤I/O速率

    Fig. 5 The network I/O rate of different jobs on node3圖5 普通節(jié)點(diǎn)node3上不同作業(yè)的網(wǎng)絡(luò)I/O速率

    同樣,對(duì)于內(nèi)存利用率、磁盤I/O和網(wǎng)絡(luò)I/O的情況,在作業(yè)類型變化時(shí)也具有不同的特點(diǎn).圖3~5分別為本文提出的CCM策略下監(jiān)控普通節(jié)點(diǎn)node3在執(zhí)行過(guò)程中的內(nèi)存利用率、磁盤I/O速率和網(wǎng)絡(luò)I/O速率變化情況.在內(nèi)存利用率上,與作業(yè)的類型和輸入數(shù)據(jù)的分布情況有關(guān).對(duì)于相同的算法而言,所需處理的數(shù)據(jù)量越大,內(nèi)存資源使用量越大.由圖3可知,WordCount和TeraSort隨著執(zhí)行時(shí)間的增加,具有相對(duì)穩(wěn)定的內(nèi)存占用率;而K-Means和PageRank則隨著處理任務(wù)的階段不同,具有不同的內(nèi)存占用率.

    在磁盤I/O速率方面,無(wú)論任務(wù)是處理本地?cái)?shù)據(jù)還是網(wǎng)絡(luò)數(shù)據(jù),都會(huì)在某個(gè)節(jié)點(diǎn)上產(chǎn)生相應(yīng)的本地?cái)?shù)據(jù)讀取,消耗一定的磁盤I/O。若處理網(wǎng)絡(luò)數(shù)據(jù),還要產(chǎn)生額外的網(wǎng)絡(luò)I/O.由于作業(yè)需要從磁盤中讀取數(shù)據(jù),并在新的RDD生成時(shí)設(shè)置檢查點(diǎn),因此產(chǎn)生了較為頻繁的磁盤I/O.由圖4可知,其中WordCount的磁盤I/O更為明顯,其他3類作業(yè)則頻率較低.K-Means在100 s后磁盤I/O有明顯增加,這是由于節(jié)點(diǎn)失效,分配給node3恢復(fù)任務(wù),此時(shí)需要從磁盤中讀取相應(yīng)檢查點(diǎn),從而實(shí)現(xiàn)恢復(fù).

    對(duì)于網(wǎng)絡(luò)I/O而言,其開銷大小與作業(yè)的類型和任務(wù)并行度有關(guān).由于Spark具有數(shù)據(jù)本地性的特點(diǎn),盡可能保證節(jié)點(diǎn)上執(zhí)行的任務(wù)處理本地?cái)?shù)據(jù),因此網(wǎng)絡(luò)I/O主要源于寬依賴階段,另外小部分網(wǎng)絡(luò)I/O則是處理遠(yuǎn)程數(shù)據(jù).由圖5可看出,對(duì)于4類不同的作業(yè),網(wǎng)絡(luò)I/O開銷相差不大.其中WordCount作業(yè)的寬依賴階段只需合并少量數(shù)據(jù),因此其網(wǎng)絡(luò)I/O開銷較小.

    另外,由表2可知,在設(shè)置檢查點(diǎn)的情況下,本文提出的CCM策略的K-Means和PageRank的恢復(fù)加速情況要好于Tachyon的檢查點(diǎn)策略,而WordCount和TeraSort與Tachyon類似,這是對(duì)比K-Means和PageRank,WordCount和TeraSort中的寬依賴操作較少,因此在使用基于lineage的檢查點(diǎn)策略也能夠較好地設(shè)置失效恢復(fù)所需檢查點(diǎn).并且計(jì)算開銷與算法復(fù)雜度和輸入數(shù)據(jù)的大小相關(guān),當(dāng)作業(yè)一定時(shí)輸入數(shù)據(jù)越大,算法開銷越大,在此粗略地認(rèn)為輸入數(shù)據(jù)越大,相應(yīng)RDD的計(jì)算開銷也會(huì)相應(yīng)增加,同時(shí)增加磁盤讀寫的開銷.對(duì)應(yīng)不同的作業(yè),節(jié)點(diǎn)的處理速度也不同.

    4.3 檢查點(diǎn)設(shè)置算法

    為了進(jìn)一步對(duì)比和分析CCM策略,我們選用PageRank進(jìn)行性能測(cè)試、評(píng)價(jià)與比較.實(shí)驗(yàn)數(shù)據(jù)選用SNAP[25]提供的有向圖數(shù)據(jù)集,數(shù)據(jù)集列表如表3所示.

    利用節(jié)點(diǎn)和連接數(shù)差異較大的2個(gè)數(shù)據(jù)集Web-Google和Wiki-Talk分別迭代1~10次對(duì)該算法性能驗(yàn)證,并使用nmon監(jiān)測(cè)執(zhí)行時(shí)間和任務(wù)檢查點(diǎn)的大小.PageRank任務(wù)在多個(gè)數(shù)據(jù)集上執(zhí)行,使用PageRank有2方面的原因:1)PageRank主要用于有向圖的計(jì)算,是一個(gè)典型的迭代計(jì)算算法;2)PageRank是計(jì)算密集型算法,因此對(duì)檢查點(diǎn)系統(tǒng)更敏感,更利于驗(yàn)證算法.對(duì)于數(shù)據(jù)密集型任務(wù)而言,代價(jià)較高,因?yàn)樾柙趲掃h(yuǎn)低于內(nèi)存的集群網(wǎng)絡(luò)間拷貝大量的數(shù)據(jù),同時(shí)也將產(chǎn)生大量的存儲(chǔ)開銷.

    Table 3 Information of Datasets表3 測(cè)試數(shù)據(jù)集列表

    圖6表示與現(xiàn)有Spark內(nèi)存系統(tǒng)使用不同策略的檢查點(diǎn)設(shè)置算法的執(zhí)行效率進(jìn)行對(duì)比的情況.其中Web-Google和Wiki-Talk帶下標(biāo)Tachyon,experience,CCM分別代表:1)Tachyon設(shè)置檢查點(diǎn)的策略,即僅考慮RDD的深度,即當(dāng)選擇檢查點(diǎn)時(shí),選擇最新生成的RDD進(jìn)行持久化存儲(chǔ);2)Spark程序員選取檢查點(diǎn)的經(jīng)驗(yàn),僅考慮RDD的操作復(fù)雜度,即當(dāng)設(shè)置檢查點(diǎn)時(shí)選擇寬依賴的RDD進(jìn)行持久化存儲(chǔ);3)采用本文提出的基于關(guān)鍵度的檢查點(diǎn)設(shè)置算法,即當(dāng)選擇檢查點(diǎn)時(shí)選擇最新生成的RDD中關(guān)鍵度最大的進(jìn)行持久化存儲(chǔ).

    Fig. 6 Efficiency of checkpoint algorithm with different parameters圖6 不同策略下檢查點(diǎn)設(shè)置算法的執(zhí)行效率

    由圖6可知,對(duì)比不同數(shù)據(jù)集,Wiki-Talk具有較大的連接數(shù)和節(jié)點(diǎn)數(shù),計(jì)算代價(jià)較高,因此無(wú)論采取什么樣的參數(shù)設(shè)置,都使其具有較大的檢查點(diǎn)大小,因此執(zhí)行時(shí)間較長(zhǎng),對(duì)應(yīng)檢查點(diǎn)存儲(chǔ)設(shè)置的平均時(shí)間開銷也隨之增加,并且迭代次數(shù)的增加對(duì)檢查點(diǎn)平均時(shí)間開銷的影響較小.由于全局檢查點(diǎn)備選列表是在計(jì)算之前生成,因此對(duì)任務(wù)實(shí)際時(shí)間影響較小.隨著任務(wù)和輸入數(shù)據(jù)的規(guī)模不同,設(shè)置時(shí)間不同.檢查點(diǎn)選擇的大小,平均時(shí)間與所選的數(shù)據(jù)集有關(guān).

    而對(duì)比不同的參數(shù)情況,與其他情況相比,僅考慮RDD計(jì)算代價(jià)的情況下設(shè)置的檢查點(diǎn)時(shí)間開銷更大,因?yàn)橛?jì)算代價(jià)與RDD的大小和復(fù)雜度都相關(guān),因此具有更大的檢查點(diǎn)大小.雖然可以用于宕機(jī)后RDD的恢復(fù),但對(duì)比很長(zhǎng)血統(tǒng)的RDD來(lái)說(shuō)這樣的恢復(fù)耗時(shí)較長(zhǎng).因此,將某些RDD進(jìn)行檢查點(diǎn)操作保存在穩(wěn)定存儲(chǔ)上是有幫助的.通常情況下,對(duì)于包含寬依賴的長(zhǎng)血統(tǒng)的RDD設(shè)置檢查點(diǎn)操作是非常有用的.比如PageRank算法中的排名數(shù)據(jù)集.在這種情況下,集群中某個(gè)節(jié)點(diǎn)的故障會(huì)使從各個(gè)父RDD得出某些數(shù)據(jù)丟失,這時(shí)候就需要重算.相反,對(duì)于那些窄依賴與穩(wěn)定存儲(chǔ)上數(shù)據(jù)的父RDD來(lái)說(shuō),對(duì)其進(jìn)行檢查點(diǎn)操作就不是必要的.

    4.4 檢查點(diǎn)恢復(fù)算法

    通過(guò)實(shí)驗(yàn)在節(jié)點(diǎn)的失效率(failure rate,fr)分別取值為0.125,0.25,0.375的情況下,對(duì)比Web-Google和Wiki-Talk數(shù)據(jù)集得到PageRank在不同恢復(fù)算法下的執(zhí)行時(shí)間和恢復(fù)情況.

    由圖7可知,對(duì)比失效率不同的情況,隨著失效節(jié)點(diǎn)個(gè)數(shù)的增加,任務(wù)執(zhí)行時(shí)間也隨之增加.這是由于失效率越大,意味著失效的節(jié)點(diǎn)越多,因此要恢復(fù)的RDD越多,需要重新計(jì)算產(chǎn)生相應(yīng)的時(shí)間開銷.對(duì)比不同數(shù)據(jù)集的情況,Wiki-Talk和Web-Google在不同算法下Wiki-Talk的時(shí)間開銷差距較大,這是由于計(jì)算量大小的區(qū)別.

    Fig. 7 Execution time of different datasets with different fr圖7 不同失效率時(shí)不同數(shù)據(jù)集的執(zhí)行時(shí)間對(duì)比

    對(duì)迭代次數(shù)1~10進(jìn)行對(duì)比,由于未設(shè)置檢查點(diǎn)的Spark原生系統(tǒng)需要利用血統(tǒng)進(jìn)行恢復(fù)丟失的RDD,因此當(dāng)?shù)螖?shù)增加時(shí)時(shí)間開銷隨之增加.在節(jié)點(diǎn)失效時(shí),同時(shí)會(huì)使運(yùn)行在該節(jié)點(diǎn)的任務(wù)執(zhí)行失敗,同時(shí)丟失存儲(chǔ)在其內(nèi)存上的已生成RDD分區(qū).在失效恢復(fù)時(shí),Spark為了對(duì)丟失的RDD分區(qū)進(jìn)行恢復(fù),利用集群中其他主機(jī)重新執(zhí)行失效任務(wù).這些任務(wù)需要將輸入數(shù)據(jù)重新讀取,并利用血統(tǒng)進(jìn)行RDD重建.隨著作業(yè)迭代次數(shù)的增加,血統(tǒng)變長(zhǎng),此時(shí)需要計(jì)算RDD的時(shí)間越長(zhǎng),則恢復(fù)開銷越大.而失效恢復(fù)策略在恢復(fù)時(shí)間開銷方面沒(méi)有顯著的增加,因?yàn)樵摬呗岳脵z查點(diǎn)設(shè)置算法設(shè)置了檢查點(diǎn),可以通過(guò)從HDFS中讀取檢查點(diǎn)進(jìn)行恢復(fù),從而減少RDD的重復(fù)計(jì)算,降低恢復(fù)時(shí)間開銷.

    4.5 檢查點(diǎn)清理算法

    定義6. 有效空間利用率.在群集工作結(jié)點(diǎn)的備份所有檢查點(diǎn)RDD之中,對(duì)恢復(fù)執(zhí)行效率有加速作用的容量之和占檢查點(diǎn)總?cè)萘康谋嚷?

    檢查點(diǎn)清理(checkpoint cleaning, CC)算法是檢查點(diǎn)選擇算法的后續(xù)同步操作,同樣在磁盤趨近滿載時(shí)才能體現(xiàn)效能.圖8為對(duì)比Web-Google和Wiki-Talk數(shù)據(jù)集在使用檢查點(diǎn)清理算法策略時(shí)和未優(yōu)化時(shí)的有效空間利用率.如圖8所示,傳統(tǒng)Spark框架隨著并發(fā)應(yīng)用數(shù)的增加,有效磁盤利用率的惡化程度也越來(lái)越高;而檢查點(diǎn)清理算法的有效磁盤利用率則較為穩(wěn)定.對(duì)比來(lái)看,采用檢查點(diǎn)清理算法的Spark框架,其有效磁盤利用率普遍高于傳統(tǒng)Spark框架.根據(jù)并行任務(wù)的特性,當(dāng)其父RDD的所有子RDD都被設(shè)置檢查點(diǎn)時(shí),則該父RDD在恢復(fù)時(shí)不會(huì)被使用到,因此存儲(chǔ)該RDD對(duì)作業(yè)恢復(fù)效率沒(méi)有影響.而檢查點(diǎn)清理算法在工作節(jié)點(diǎn)的某個(gè)RDD需要清理時(shí),通知群集的其他工作節(jié)點(diǎn)清除該RDD的其他副本,因此能在不影響任務(wù)恢復(fù)效率的前提下提高了群集磁盤空間的有效利用率.

    Fig. 8 Valuable capacity rate of CC圖8 檢查點(diǎn)清理算法的有效空間利用率

    綜合比較表2以及圖3~8,在不同數(shù)據(jù)集、策略的情況下,對(duì)比任務(wù)的執(zhí)行時(shí)間和恢復(fù)時(shí)間可知,在提出失效恢復(fù)策略中,檢查點(diǎn)策略會(huì)增加少量的時(shí)間開銷,然而對(duì)比傳統(tǒng)Spark策略中,由程序員選擇檢查點(diǎn)的不確定因素甚至是異常風(fēng)險(xiǎn),這些額外時(shí)間和空間開銷是有價(jià)值的.失效恢復(fù)算法基于檢查點(diǎn)設(shè)置算法,在設(shè)置檢查點(diǎn)時(shí)考慮RDD的血統(tǒng)長(zhǎng)度、計(jì)算代價(jià)、操作復(fù)雜度和容量等因素.權(quán)重越大的RDD,重新計(jì)算的恢復(fù)成本也越高,優(yōu)先將這些恢復(fù)成本較高的RDD設(shè)置為檢查點(diǎn),可以降低任務(wù)整體的重新計(jì)算代價(jià).在執(zhí)行恢復(fù)算法時(shí),選擇計(jì)算能力強(qiáng)的節(jié)點(diǎn)進(jìn)行恢復(fù),可以對(duì)任務(wù)的恢復(fù)效率進(jìn)行有效的提高.

    5 總結(jié)與展望

    本文針對(duì)內(nèi)存計(jì)算框架Spark的失效恢復(fù)問(wèn)題,首先對(duì)內(nèi)存計(jì)算框架的任務(wù)執(zhí)行機(jī)制進(jìn)行分析,建立執(zhí)行模型和檢查點(diǎn)模型.通過(guò)分析檢查點(diǎn)恢復(fù)過(guò)程,給出了RDD關(guān)鍵度和失效恢復(fù)比的定義,并證明這些定義與任務(wù)恢復(fù)效率的關(guān)系,為算法設(shè)計(jì)提供基礎(chǔ)模型.在相關(guān)模型定義和證明的基礎(chǔ)上,提出了基于RDD關(guān)鍵度的檢查點(diǎn)管理策略問(wèn)題定義,以此作為算法設(shè)計(jì)的主要依據(jù).通過(guò)算法的問(wèn)題定義求解,計(jì)算RDD關(guān)鍵度,設(shè)計(jì)了檢查點(diǎn)設(shè)置算法、恢復(fù)算法和并行清理算法.最后,通過(guò)不同的實(shí)驗(yàn)證明算法的有效性,實(shí)驗(yàn)結(jié)果表明,該策略優(yōu)化了內(nèi)存計(jì)算框架的檢查點(diǎn)管理,提高了作業(yè)的恢復(fù)效率.

    未來(lái)工作主要集中在3個(gè)方面:

    1) 分析內(nèi)存計(jì)算框架不同類型操作資源需求的一般規(guī)律,設(shè)計(jì)適應(yīng)作業(yè)負(fù)載和類型的檢查點(diǎn)策略;

    2) 對(duì)內(nèi)存計(jì)算框架的內(nèi)存管理進(jìn)行研究,優(yōu)化Executor現(xiàn)有的任務(wù)內(nèi)存分配策略;

    3) 通過(guò)分析計(jì)算節(jié)點(diǎn)性能和作業(yè)DAG圖,利用樣本和歷史記錄執(zhí)行預(yù)測(cè)RDD和作業(yè)的計(jì)算開銷,從而協(xié)助資源分配和并行任務(wù)調(diào)度方案做決策.

    [1]Walker S J. Big data: A revolution that will transform how we live, work, and think[J]. International Journal of Advertising, 2014, 17(1): 181-183

    [2] Meng Xiaofeng, Ci Xiang. Big data management: Concepts, techniques and challenges[J]. Journal of Computer Research and Development, 2013, 50(1): 146-169 (in Chinese)(孟小峰, 慈祥. 大數(shù)據(jù)管理: 概念、技術(shù)與挑戰(zhàn)[J]. 計(jì)算機(jī)研究與發(fā)展, 2013, 50(1): 146-169)

    [3] Chen C P, Zhang Chunyang. Data-intensive applications, challenges, techniques and technologies: A survey on big data[J]. Information Sciences, 2014, 275(11): 314-347

    [4] Kambatla K, Kollias G, Kumar V, et al. Trends in big data analytics[J]. Journal of Parallel and Distributed Computing, 2014, 74(7): 2561-2573

    [5] Zaharia M, Chowdhury M, Das T, et al. Fast and interactive analytics over Hadoop data with Spark[J]. ;Login:, 2012, 37(4): 45-51

    [6] Apache. Spark overview[EB/OL]. 2011 [2016-03-18]. http://spark.apache.org

    [7] 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

    [8] SAP. HANA overview[EB/OL]. 2011 [2016-09-21]. http://hana.sap.com/abouthana.html

    [9] Dean J, Ghemawat S. MapReduce: Simplified data processing on large clusters[C] //Proc of the 6th Symp on Operating System Design and Implementation (OSDI). New York: ACM, 2004: 137-150

    [10] Apache. Spark machine learning library (MLlib)[EB/OL]. 2012 [2016-03-18]. http://spark.incubator.apache.org/docs/latest/mllib-guide.html

    [11] Lin Xiuqin, Wang Peng, Wu Bin. Log analysis in cloud computing environment with Hadoop and Spark[C] //Proc of the 5th IEEE Int Conf on Broadband Network & Multimedia Technology (IC-BNMT). Piscataway, NJ: IEEE, 2013: 273-276

    [12] Dong Xiangyu, Xie Yuan, Muralimanohar N, et al. Hybrid checkpointing using emerging nonvolatile memories for future exascale system[J]. ACM Trans on Architecture and Code Optimization, 2011, 8(2): Article No.6

    [13] Dimitriou I. A retrial queue for modeling fault-tolerant systems with checkpointing and rollback recovery[J]. Computers & Industrial Engineering, 2015, 79: 156-167

    [14] Ifeanyi P E, David L, Bran S, et al. A survey of fault tolerance mechanisms and checkpoint/restart implementations for high performance computing systems[J]. Journal of Supercomputing, 2013, 65(3): 1302-1326

    [15] Zhou Enqiang, Lu Yutong, Shen Zhiyu. Implementation of checkpoint system toward large scale parallel computing[J]. Journal of Computer Research and Development, 2005, 42(6): 987-992 (in Chinese)(周恩強(qiáng), 盧宇彤, 沈志宇. 一個(gè)適合大規(guī)模集群并行計(jì)算的檢查點(diǎn)系統(tǒng)[J]. 計(jì)算機(jī)研究與發(fā)展, 2005, 42(6): 987-992)

    [16] Yi Huizhan, Wang Feng, Zuo Ke, et al. Asynchronous checkpoint/restart based on memory buffer[J]. Journal of Computer Research and Development, 2014, 51(6): 1229-1239 (in Chinese)(易會(huì)戰(zhàn), 王鋒, 左克, 等. 基于內(nèi)存緩存的異步檢查點(diǎn)容錯(cuò)技術(shù)[J].計(jì)算機(jī)研究與發(fā)展, 2014, 51(6): 1229-1239)

    [17] Wan Hu, Xu Yuanchao, Yan Junfeng, et al. Mitigating log cost through non-volatile memory and checkpoint optimization[J]. Journal of Computer Research and Development, 2015, 52(6): 1351-1361 (in Chinese)(萬(wàn)虎, 徐遠(yuǎn)超, 閆俊峰, 等. 通過(guò)非易失存儲(chǔ)和檢查點(diǎn)優(yōu)化緩解日志開銷[J]. 計(jì)算機(jī)研究與發(fā)展, 2015, 52(6): 1351-1361)

    [18] Cores I, Rodríguez G, Martín M J, et al. In-memory application-level checkpoint-based migration for MPI programs[J]. Journal of Supercomputing, 2014, 70(2): 660-670

    [19] Cao T, Vaz S M, Sowell B, et al. Fast checkpoint recovery algorithms for frequently consistent applications[C] //Proc of the 2011 ACM SIGMOD Int Conf on Management of Data. New York: ACM, 2011: 265-276

    [20] Chardonnens T, Cudre-Mauroux P, Grund M, et al. Big data analytics on high velocity streams: A case study[C] //Proc of 2013 IEEE Int Conf on Big Data. Piscataway, NJ: IEEE, 2013: 784-787

    [21] Neumeyer L, Robbins B, Nair A, et al. S4: Distributed stream computing platform[C] //Proc of the 10th IEEE Int Conf on Data Mining Workshops (ICDMW 2010). Piscataway, NJ: IEEE, 2010: 170-177

    [22] Li Haoyuan, Ghodsi A, Zaharia M, et al. Tachyon: Memory throughput I/O for cluster computing frameworks[C/OL]. 2013 [2016-03-18]. https://people.eecs.berkeley.edu/~alig/papers/tachyon-workshop.pdf

    [23] Li Haoyuan, Ghodsi A, Zaharia M, et al. Tachyon: Reliable, memory speed storage for cluster computing frameworks[C] //Proc of the 2014 ACM Symp on Cloud Computing. New York: ACM, 2014

    [24] Ongaro D, Rumble S M, Stutsman R, et al. Fast crash recovery in RAMCloud[C] //Proc of the 23rd ACM Symp on Operating Systems Principles. New York: ACM, 2011: 29-41

    [25] Jure L. Stanford network analysis project[EB/OL]. 2009 [2016-03-18]. http://snap.stanford.edu

    CriticalityCheckpointManagementStrategyBasedonRDDCharacteristicsinSpark

    Ying Changtian1,2, Yu Jiong2, Bian Chen2, Wang Weiqing1,3, Lu Liang2, and Qian Yurong2

    1(PostdoctoralResearchStationofElectricalEngineering,XinjiangUniversity,Urumqi830046)2(SchoolofSoftware,XinjiangUniversity,Urumqi830008)3(SchoolofElectricalEngineering,XinjiangUniversity,Urumqi830046)

    The default fault tolerance mechanism of Spark is setting the checkpoint by programmer. When facing data loss, Spark recomputes the tasks based on the RDD lineage to recovery the data. Meanwhile, in the circumstance of complicated application with multiple iterations and large amount of input data, the recovery process may cost a lot of computation time. In addition, the recompute task only considers the data locality by default regardless the computing capabilities of nodes, which increases the length of recovery time. To reduce recovery cost, we establish and demonstrate the Spark execution model, the checkpoint model and the RDD critically model. Based on the theory, the criticality checkpoint management (CCM) strategy is proposed, which includes the checkpoint algorithm, the failure recovery algorithm and the cleaning algorithm. The checkpoint algorithm is used to analyze the RDD charactersitics and its influence on the recovery time, and selects valuable RDDs as checkpoints. The failure recovery algorithm is used to choose the appropriate nodes to recompute the lost RDDs, and cleaning algorithm cleans checkpoints when the disk space becomes insufficient. Experimental results show that: the strategy can reduce the recovery overhead efficiently, select valuable RDDs as checkpoints, and increase the efficiency of disk usage on the nodes with sacrificing the execution time slightly.

    memory computing; Spark; checkpoint management; failure recovery; RDD characteristics

    2016-09-20;

    2017-07-04

    國(guó)家自然科學(xué)基金項(xiàng)目(61262088,61462079,61363083,61562086,51667020);新疆維吾爾自治區(qū)自然科學(xué)基金項(xiàng)目(2017D01A20);新疆維吾爾自治區(qū)高??蒲杏?jì)劃(XJEDU2016S106)

    This work was supported by the National Natural Science Foundation of China (61262088, 61462079, 61363083, 61562086, 51667020), the Natural Science Foundation of Xinjiang Uygur Autonomous Region of China (2017D01A20), and the Higher Education Research Program of Xinjiang Uygur Autonomous Region (XJEDU2016S106).

    于炯(yujiong@xju.edu.cn)

    TP311

    YingChangtian, born in 1989. PhD in Xinjiang University. Student member of CCF. Her main research interests include parallel computing, distributed system, and memory computing, etc.

    YuJiong, born in 1964. Professor and PhD supervisor. Senior member of CCF. His main research interests include grid computing, parallel computing, etc.

    BianChen, born in 1981. Associate professor and PhD. Senior member of CCF. His main research interests include parallel computing, distributed system, etc.

    WangWeiqing, born in 1959. Professor and PhD supervisor. His main research interests include power system relay protection, wind power generation control and grid connection technology (wwq59@xju.edu.cn).

    LuLiang, born in 1990. PhD candidate in Xinjiang University. Student member of CCF. His main research interests include flow processing, real-time computing.

    QianYurong, born in 1981. Professor and master supervisor. Senior member of CCF. Her main research interests include data mining.

    猜你喜歡
    檢查點(diǎn)內(nèi)存分區(qū)
    Spark效用感知的檢查點(diǎn)緩存并行清理策略①
    上海實(shí)施“分區(qū)封控”
    免疫檢查點(diǎn)抑制劑相關(guān)內(nèi)分泌代謝疾病
    “春夏秋冬”的內(nèi)存
    免疫檢查點(diǎn)抑制劑在腫瘤治療中的不良反應(yīng)及毒性管理
    浪莎 分區(qū)而治
    分布式任務(wù)管理系統(tǒng)中檢查點(diǎn)的設(shè)計(jì)
    基于SAGA聚類分析的無(wú)功電壓控制分區(qū)
    基于多種群遺傳改進(jìn)FCM的無(wú)功/電壓控制分區(qū)
    基于內(nèi)存的地理信息訪問(wèn)技術(shù)
    久久久久久久久中文| 每晚都被弄得嗷嗷叫到高潮| 欧美激情在线99| 国产精品美女特级片免费视频播放器| 少妇的逼好多水| 丰满乱子伦码专区| 亚洲自偷自拍三级| 精品久久久久久久久av| 国产私拍福利视频在线观看| 男女做爰动态图高潮gif福利片| 国内少妇人妻偷人精品xxx网站| 日本免费一区二区三区高清不卡| 麻豆久久精品国产亚洲av| av在线蜜桃| 久久精品国产自在天天线| or卡值多少钱| 成年免费大片在线观看| 色综合婷婷激情| 悠悠久久av| 国产伦精品一区二区三区视频9| 精品国产三级普通话版| 成人毛片a级毛片在线播放| 成人特级av手机在线观看| 欧美一区二区国产精品久久精品| 亚洲精品色激情综合| 一区二区三区四区激情视频 | 亚洲国产精品合色在线| 一a级毛片在线观看| 一级毛片久久久久久久久女| 久久精品国产99精品国产亚洲性色| 少妇的逼水好多| 国产精品一区二区性色av| 国产欧美日韩一区二区三| 女人十人毛片免费观看3o分钟| 国产日本99.免费观看| 成人一区二区视频在线观看| 久久久久久九九精品二区国产| 97超视频在线观看视频| 好男人在线观看高清免费视频| 99久久精品国产亚洲精品| 可以在线观看毛片的网站| 亚洲人成网站在线播放欧美日韩| 免费av毛片视频| 日日干狠狠操夜夜爽| 老司机午夜福利在线观看视频| 欧美+日韩+精品| 午夜影院日韩av| 午夜福利免费观看在线| 老司机午夜福利在线观看视频| 亚洲午夜理论影院| 日本与韩国留学比较| 国产国拍精品亚洲av在线观看| 午夜两性在线视频| 男女床上黄色一级片免费看| 亚洲av电影在线进入| 亚洲国产色片| 日日摸夜夜添夜夜添av毛片 | 久久久精品欧美日韩精品| 18禁黄网站禁片午夜丰满| 日日摸夜夜添夜夜添小说| 国产精品综合久久久久久久免费| 亚洲性夜色夜夜综合| 午夜激情福利司机影院| 日韩欧美在线乱码| 久久国产乱子免费精品| 老鸭窝网址在线观看| 亚洲美女搞黄在线观看 | 免费av观看视频| 亚洲精品在线观看二区| 国产高清有码在线观看视频| 天堂影院成人在线观看| 午夜福利在线观看免费完整高清在 | 国产精品日韩av在线免费观看| 欧美日韩福利视频一区二区| 欧美+日韩+精品| 亚洲成人中文字幕在线播放| 白带黄色成豆腐渣| 丰满人妻一区二区三区视频av| 十八禁人妻一区二区| 国产av一区在线观看免费| 欧美最新免费一区二区三区 | 亚洲第一欧美日韩一区二区三区| 日日摸夜夜添夜夜添av毛片 | 岛国在线免费视频观看| 乱码一卡2卡4卡精品| 一个人看的www免费观看视频| 久久精品国产亚洲av香蕉五月| 男女视频在线观看网站免费| 观看免费一级毛片| 久久精品人妻少妇| 亚洲性夜色夜夜综合| xxxwww97欧美| 伦理电影大哥的女人| av欧美777| 日韩高清综合在线| 怎么达到女性高潮| 国产精品综合久久久久久久免费| 一夜夜www| 亚洲男人的天堂狠狠| 欧美最新免费一区二区三区 | 老司机福利观看| 亚洲欧美日韩东京热| 国产一区二区在线观看日韩| 国产亚洲av嫩草精品影院| 国产色婷婷99| 日本与韩国留学比较| 久久6这里有精品| 中文字幕免费在线视频6| 亚洲av五月六月丁香网| 日韩欧美三级三区| 一夜夜www| 免费大片18禁| 日韩精品中文字幕看吧| 黄色一级大片看看| 亚洲久久久久久中文字幕| 又爽又黄a免费视频| 中文在线观看免费www的网站| 成人av在线播放网站| 国产人妻一区二区三区在| 午夜福利视频1000在线观看| 99热这里只有是精品在线观看 | 麻豆久久精品国产亚洲av| 男人舔女人下体高潮全视频| 国产欧美日韩精品亚洲av| 色播亚洲综合网| 精品无人区乱码1区二区| 日韩大尺度精品在线看网址| 精品久久久久久,| 日韩av在线大香蕉| 内地一区二区视频在线| 久久人妻av系列| 亚洲美女视频黄频| 午夜a级毛片| 五月玫瑰六月丁香| 中文字幕熟女人妻在线| 久久6这里有精品| 国产麻豆成人av免费视频| 在现免费观看毛片| 99热6这里只有精品| 99久久精品国产亚洲精品| 赤兔流量卡办理| 免费无遮挡裸体视频| 亚洲美女黄片视频| 亚洲熟妇中文字幕五十中出| 国内毛片毛片毛片毛片毛片| 久久性视频一级片| 国产亚洲精品综合一区在线观看| 久久精品国产亚洲av香蕉五月| 日韩亚洲欧美综合| 亚洲va日本ⅴa欧美va伊人久久| 九九久久精品国产亚洲av麻豆| 成年人黄色毛片网站| 欧美最黄视频在线播放免费| 日韩成人在线观看一区二区三区| 国产亚洲精品久久久com| 人人妻人人看人人澡| 美女免费视频网站| 久久久国产成人免费| 在线播放无遮挡| 在线观看一区二区三区| 精品一区二区三区视频在线| 老司机深夜福利视频在线观看| 亚洲美女视频黄频| 欧美日韩福利视频一区二区| 在线十欧美十亚洲十日本专区| 搡女人真爽免费视频火全软件 | 亚洲国产高清在线一区二区三| 亚洲精品粉嫩美女一区| 国产成人啪精品午夜网站| 欧美日韩黄片免| 亚洲最大成人av| 国产成人啪精品午夜网站| 人妻夜夜爽99麻豆av| 一级毛片久久久久久久久女| 最近最新中文字幕大全电影3| 日本免费一区二区三区高清不卡| 一区二区三区免费毛片| 中文资源天堂在线| 十八禁网站免费在线| 日本撒尿小便嘘嘘汇集6| 国产黄片美女视频| 国产一区二区在线av高清观看| 中出人妻视频一区二区| 中文字幕av成人在线电影| 一区二区三区四区激情视频 | 国产一区二区激情短视频| 国产视频内射| 全区人妻精品视频| 我要看日韩黄色一级片| 老师上课跳d突然被开到最大视频 久久午夜综合久久蜜桃 | 欧美黑人巨大hd| 乱码一卡2卡4卡精品| 91字幕亚洲| 午夜亚洲福利在线播放| 国产精品一区二区免费欧美| 美女免费视频网站| 少妇人妻精品综合一区二区 | 超碰av人人做人人爽久久| 天堂影院成人在线观看| 人妻久久中文字幕网| 不卡一级毛片| 亚洲黑人精品在线| 国产精品人妻久久久久久| 男女做爰动态图高潮gif福利片| 国产人妻一区二区三区在| 久久久久国产精品人妻aⅴ院| 美女黄网站色视频| 色av中文字幕| 久久久久亚洲av毛片大全| 男女之事视频高清在线观看| 国产精品野战在线观看| 国产精品免费一区二区三区在线| 免费看日本二区| 99视频精品全部免费 在线| АⅤ资源中文在线天堂| 黄色丝袜av网址大全| 国产三级中文精品| 久久亚洲真实| 啦啦啦韩国在线观看视频| 午夜福利欧美成人| bbb黄色大片| 在线十欧美十亚洲十日本专区| 久久精品影院6| 久久99热6这里只有精品| 精品不卡国产一区二区三区| 欧美日韩乱码在线| 成人精品一区二区免费| 他把我摸到了高潮在线观看| 国产成人欧美在线观看| av天堂中文字幕网| а√天堂www在线а√下载| 夜夜躁狠狠躁天天躁| 国产免费一级a男人的天堂| 亚洲精品粉嫩美女一区| 久久久久久久亚洲中文字幕 | 亚洲第一电影网av| 午夜影院日韩av| 一本一本综合久久| 小说图片视频综合网站| 国产精品久久久久久人妻精品电影| 免费一级毛片在线播放高清视频| 日本 av在线| 免费在线观看成人毛片| 国内少妇人妻偷人精品xxx网站| 色精品久久人妻99蜜桃| 亚洲最大成人中文| 黄色视频,在线免费观看| 成人美女网站在线观看视频| 他把我摸到了高潮在线观看| 欧美极品一区二区三区四区| 亚洲真实伦在线观看| 成年人黄色毛片网站| 国产精品一区二区三区四区免费观看 | 国产精品亚洲一级av第二区| 欧美在线一区亚洲| 内射极品少妇av片p| 久久人人爽人人爽人人片va | 国产男靠女视频免费网站| 亚洲五月婷婷丁香| 欧美日韩综合久久久久久 | 久久久久九九精品影院| 欧美激情国产日韩精品一区| 亚洲狠狠婷婷综合久久图片| 国产精品野战在线观看| АⅤ资源中文在线天堂| 久久久精品大字幕| 国产精品av视频在线免费观看| 中文字幕精品亚洲无线码一区| 欧美最新免费一区二区三区 | 国产成人影院久久av| 狠狠狠狠99中文字幕| 日韩欧美精品免费久久 | 久久国产乱子伦精品免费另类| 国产国拍精品亚洲av在线观看| 别揉我奶头~嗯~啊~动态视频| 国产av麻豆久久久久久久| 亚洲中文日韩欧美视频| 精品久久久久久久久亚洲 | 岛国在线免费视频观看| 人妻久久中文字幕网| 一本综合久久免费| av天堂在线播放| 嫩草影院新地址| 色综合欧美亚洲国产小说| 国产伦一二天堂av在线观看| 欧美日本亚洲视频在线播放| 超碰av人人做人人爽久久| 嫁个100分男人电影在线观看| 人人妻人人看人人澡| 九九久久精品国产亚洲av麻豆| 热99re8久久精品国产| 人妻制服诱惑在线中文字幕| 一a级毛片在线观看| 给我免费播放毛片高清在线观看| 国产精品久久电影中文字幕| 毛片女人毛片| 看免费av毛片| 好男人电影高清在线观看| 男女之事视频高清在线观看| 日韩欧美在线二视频| 99国产精品一区二区三区| 日本 av在线| 别揉我奶头~嗯~啊~动态视频| 特级一级黄色大片| av在线观看视频网站免费| 成人午夜高清在线视频| 国产成人av教育| 精品人妻一区二区三区麻豆 | 97超视频在线观看视频| 国产高清有码在线观看视频| 国产成人啪精品午夜网站| 麻豆久久精品国产亚洲av| 最近视频中文字幕2019在线8| 久久精品国产99精品国产亚洲性色| h日本视频在线播放| 我的女老师完整版在线观看| 三级毛片av免费| 国产精品乱码一区二三区的特点| 欧美午夜高清在线| 国产真实乱freesex| 在线十欧美十亚洲十日本专区| 高清日韩中文字幕在线| 亚洲人成电影免费在线| 国产伦在线观看视频一区| 亚洲激情在线av| 国产成人av教育| 国产精品国产高清国产av| 免费高清视频大片| 亚洲一区高清亚洲精品| 小说图片视频综合网站| 亚洲国产欧美人成| 亚洲男人的天堂狠狠| 悠悠久久av| or卡值多少钱| 尤物成人国产欧美一区二区三区| 一区二区三区高清视频在线| 天美传媒精品一区二区| 天堂影院成人在线观看| av中文乱码字幕在线| 国产精品伦人一区二区| 熟妇人妻久久中文字幕3abv| 国产高清激情床上av| 国产成人aa在线观看| 757午夜福利合集在线观看| 美女黄网站色视频| 桃色一区二区三区在线观看| 亚洲专区国产一区二区| 亚洲国产欧美人成| 午夜福利成人在线免费观看| 我的老师免费观看完整版| 免费观看人在逋| 国产综合懂色| 精品久久久久久,| 禁无遮挡网站| 好看av亚洲va欧美ⅴa在| 丰满的人妻完整版| 欧美色欧美亚洲另类二区| 在线观看66精品国产| 亚洲av日韩精品久久久久久密| 日本三级黄在线观看| 久久久久久九九精品二区国产| 亚洲av第一区精品v没综合| 精品久久久久久久久久免费视频| 国产午夜精品久久久久久一区二区三区 | 亚洲avbb在线观看| 国产精品一区二区性色av| 性色av乱码一区二区三区2| 高潮久久久久久久久久久不卡| 午夜福利18| 久久九九热精品免费| 国产免费男女视频| 精品午夜福利在线看| 国产单亲对白刺激| 国产一级毛片七仙女欲春2| 在线观看美女被高潮喷水网站 | 1000部很黄的大片| 狠狠狠狠99中文字幕| 真实男女啪啪啪动态图| 少妇人妻一区二区三区视频| 啪啪无遮挡十八禁网站| 中文字幕久久专区| 宅男免费午夜| 成人鲁丝片一二三区免费| 小说图片视频综合网站| 五月伊人婷婷丁香| 国产亚洲欧美在线一区二区| 久久99热6这里只有精品| 人妻夜夜爽99麻豆av| 天天躁日日操中文字幕| 99热精品在线国产| 日本 av在线| 麻豆成人午夜福利视频| 美女被艹到高潮喷水动态| 久久精品夜夜夜夜夜久久蜜豆| 国产精品久久视频播放| 窝窝影院91人妻| 91字幕亚洲| 一夜夜www| 日韩有码中文字幕| 99精品在免费线老司机午夜| 亚洲美女黄片视频| 久久久久国内视频| 人妻丰满熟妇av一区二区三区| 国产精品久久久久久亚洲av鲁大| 欧美在线一区亚洲| 乱码一卡2卡4卡精品| 一a级毛片在线观看| 国产成人aa在线观看| 亚洲精品一卡2卡三卡4卡5卡| 亚洲一区二区三区色噜噜| 午夜福利欧美成人| 好男人在线观看高清免费视频| 国产伦在线观看视频一区| 亚洲综合色惰| 人妻久久中文字幕网| 变态另类成人亚洲欧美熟女| 偷拍熟女少妇极品色| 欧美日韩亚洲国产一区二区在线观看| 一区二区三区激情视频| 亚洲最大成人av| 亚洲男人的天堂狠狠| 成人毛片a级毛片在线播放| 嫩草影院新地址| 免费观看精品视频网站| 久久精品国产自在天天线| 国产人妻一区二区三区在| av福利片在线观看| 偷拍熟女少妇极品色| 亚洲三级黄色毛片| 波多野结衣高清作品| 成人亚洲精品av一区二区| av欧美777| 校园春色视频在线观看| 亚州av有码| 欧美最黄视频在线播放免费| 免费看美女性在线毛片视频| 日韩精品中文字幕看吧| 国产亚洲av嫩草精品影院| 亚洲成人久久性| 亚洲人成伊人成综合网2020| 搡老岳熟女国产| 日本三级黄在线观看| 国产精品免费一区二区三区在线| 看片在线看免费视频| 一个人免费在线观看的高清视频| 日韩欧美三级三区| 日韩人妻高清精品专区| or卡值多少钱| 国产欧美日韩精品一区二区| 88av欧美| 国产色爽女视频免费观看| 久久精品国产亚洲av香蕉五月| xxxwww97欧美| 日韩欧美 国产精品| 桃红色精品国产亚洲av| 91狼人影院| 精品久久国产蜜桃| av在线老鸭窝| 亚洲激情在线av| 一卡2卡三卡四卡精品乱码亚洲| 国产精品影院久久| 狂野欧美白嫩少妇大欣赏| 欧美日韩黄片免| 最后的刺客免费高清国语| 免费无遮挡裸体视频| 欧美日韩乱码在线| 国产精品人妻久久久久久| 欧美一区二区国产精品久久精品| 淫妇啪啪啪对白视频| 久久久久久久久大av| 黄色丝袜av网址大全| 日本 av在线| 欧美成人免费av一区二区三区| 少妇人妻一区二区三区视频| 老司机午夜十八禁免费视频| 国产高清视频在线播放一区| 国产真实乱freesex| 动漫黄色视频在线观看| 国模一区二区三区四区视频| 婷婷亚洲欧美| 精品久久久久久久末码| 国产人妻一区二区三区在| 人人妻,人人澡人人爽秒播| 亚洲人成网站在线播| 美女黄网站色视频| 亚洲国产日韩欧美精品在线观看| 成人午夜高清在线视频| 免费在线观看影片大全网站| 国产成人影院久久av| 人妻夜夜爽99麻豆av| 首页视频小说图片口味搜索| 国产免费av片在线观看野外av| 免费人成在线观看视频色| 一级毛片久久久久久久久女| 99视频精品全部免费 在线| 午夜精品久久久久久毛片777| 色在线成人网| www.熟女人妻精品国产| 90打野战视频偷拍视频| 国产白丝娇喘喷水9色精品| 99在线视频只有这里精品首页| av在线观看视频网站免费| 国内久久婷婷六月综合欲色啪| АⅤ资源中文在线天堂| 国内精品一区二区在线观看| 国产亚洲av嫩草精品影院| 日本与韩国留学比较| 内地一区二区视频在线| 99在线人妻在线中文字幕| 国产伦一二天堂av在线观看| 亚洲五月婷婷丁香| 亚洲精品一区av在线观看| www.熟女人妻精品国产| 国产高清视频在线观看网站| 国产午夜精品论理片| 最近最新免费中文字幕在线| 亚洲中文字幕一区二区三区有码在线看| 亚洲欧美激情综合另类| 久久精品国产亚洲av香蕉五月| 啦啦啦韩国在线观看视频| 欧美成人性av电影在线观看| 国产一级毛片七仙女欲春2| 色视频www国产| 日韩欧美一区二区三区在线观看| 精品久久久久久久人妻蜜臀av| 亚洲av熟女| 91久久精品电影网| 国产69精品久久久久777片| 亚洲综合色惰| 亚洲无线观看免费| 俺也久久电影网| 最好的美女福利视频网| av黄色大香蕉| 俄罗斯特黄特色一大片| 午夜两性在线视频| 成人鲁丝片一二三区免费| 色吧在线观看| 久久久久久久久久成人| 免费观看的影片在线观看| 日本与韩国留学比较| av黄色大香蕉| 国内毛片毛片毛片毛片毛片| 国产一区二区三区视频了| 美女高潮的动态| 国产成年人精品一区二区| 欧美日本视频| 禁无遮挡网站| 婷婷色综合大香蕉| 怎么达到女性高潮| 亚洲黑人精品在线| 在线观看av片永久免费下载| 欧美国产日韩亚洲一区| 90打野战视频偷拍视频| 午夜两性在线视频| 成年女人毛片免费观看观看9| 亚洲av免费在线观看| 日韩国内少妇激情av| 欧美成狂野欧美在线观看| www日本黄色视频网| 日日摸夜夜添夜夜添小说| 丝袜美腿在线中文| 不卡一级毛片| 亚洲人与动物交配视频| 午夜福利18| 丰满人妻熟妇乱又伦精品不卡| 我要看日韩黄色一级片| 亚洲成人免费电影在线观看| 成年女人毛片免费观看观看9| АⅤ资源中文在线天堂| 欧美乱色亚洲激情| 精品久久久久久久末码| 欧美三级亚洲精品| 日本黄色片子视频| 日本精品一区二区三区蜜桃| 搡老熟女国产l中国老女人| 一个人免费在线观看电影| 欧美最新免费一区二区三区 | 国产在线男女| 亚洲五月天丁香| 最好的美女福利视频网| 欧美一级a爱片免费观看看| 欧美一区二区亚洲| 每晚都被弄得嗷嗷叫到高潮| 精品久久久久久成人av| 亚洲无线在线观看| 女同久久另类99精品国产91| 成人鲁丝片一二三区免费| 亚洲国产欧洲综合997久久,| 国产毛片a区久久久久| 亚洲成人久久爱视频| 亚洲av.av天堂| 免费看光身美女| 国产真实乱freesex| 丰满的人妻完整版| 一本综合久久免费| 成人av在线播放网站| 亚洲乱码一区二区免费版| 精品午夜福利视频在线观看一区| 国产免费av片在线观看野外av| 男女视频在线观看网站免费| 精品久久久久久久久久久久久| 九九久久精品国产亚洲av麻豆| 亚洲av成人精品一区久久| 国产高清视频在线观看网站| 看免费av毛片| 极品教师在线免费播放| 久久精品国产亚洲av涩爱 | 精品一区二区三区视频在线| 神马国产精品三级电影在线观看| 成熟少妇高潮喷水视频| 国产精品一及| 午夜免费男女啪啪视频观看 | 国产精品综合久久久久久久免费| 美女高潮的动态| 欧美bdsm另类|