并行計算框架Spark中一種新的RDD分區(qū)權重緩存替換算法

劉 恒,譚 良,2

1(四川師范大學 計算機科學學院,成都 610101)

2(中國科學院 計算技術研究所,北京 100190)

1 引 言

隨著數(shù)據(jù)量與日俱增,人們對數(shù)據(jù)處理效率的需求也逐步增加.Spark[1]是繼Hadoop[2]之后被提出基于內存計算的可擴展的高性能并行計算框架.針對Hadoop的不足[3,4],即大量的網絡傳輸和磁盤I/O使得效率低下,Spark使用內存進行數(shù)據(jù)計算,同樣的算法在Spark中的運行速度比Hadoop快10倍～100倍[5],而且Spark廣泛的支持了多種的計算模式,包括批處理、迭代計算、交互式查詢和流處理.這使得Spark的應用越來越廣泛,使用它的公司包括雅虎、百度和騰訊.但Spark性能方面仍需提高,特別是當大量任務集中處理時,有限的內存資源會限制Spark系統(tǒng)整體性能的發(fā)揮.緩存替換是合理利用內存資源、提高任務處理性能的關鍵技術,所以針對緩存替換算法的研究便成為了熱點[6-13].

當前Spark緩存替換算法是LRU[14].該算法的缺點是在進行緩存替換時,Spark不能預測RDD對象將來的使用順序,沒有考慮到RDD計算代價和大小等影響任務執(zhí)行效率的重要因素,度量方法不夠準確.當高重用但最近未使用或重新計算耗費極高代價的RDD的分區(qū)被替換出去,應用執(zhí)行的效率會降低.針對LRU在計算框架中的表現(xiàn)不夠理想,便有了一系列改進措施被提出.如使用其他典型的替換算法包括:FIFO、LFU、LRFU、MIN來代替LRU等[9].其中權重RDD緩存替換算法是最重要的改進措施之一.在計算過程中不同的RDD,由于計算代價不同、大小不同,所以對整個計算過程的影響便不同.然而當前權重RDD緩存替換算法存在對權重值的計算結果不準確和考慮因素不全面,對計算代價缺乏量化的問題.因此,研究RDD權重緩存替換具有一定的現(xiàn)實意義.

針對上述問題,本文提出了一種新的RDD分區(qū)權重緩存替換算法——WCSRP,WCSRP綜合考慮了RDD的計算代價、使用次數(shù)、分區(qū)的大小和生命周期四大因素,并進行量化計算.相比于之前的研究,彌補了對RDD計算代價缺乏量化的缺陷,增加考慮了應用在Spark中運算時Task的locality level決定的輸入RDD位置,這時會涉及到網絡傳輸或磁盤I/O,對應用執(zhí)行時間存在影響,另外相比較于使用RDD的權重值作為是否替換的度量工具,使用分區(qū)的權重值更符合Spark底層緩存機制.這樣使得權重值的計算更加準確,提高了內存利用率和作業(yè)執(zhí)行效率.

2 相關工作

目前,國內外對Spark系統(tǒng)的緩存替換算法做出了不少研究,取得了一些的研究成果.下面我們對這些成果進行分析和總結.

對于國外,文獻[15]提出了Tachyon,一種基于內存的分布式文件系統(tǒng),能夠充分發(fā)揮分布式集群的內存特性,能夠為Spark提供存儲服務,但在Tachyon的實現(xiàn)中,替換算法還是采用LRU.文獻[10]提出了在內存空間不足的情況下,在內存上壓縮數(shù)據(jù)來節(jié)省內存空間,然而壓縮數(shù)據(jù)又會帶來CPU資源消耗的問題,在對數(shù)據(jù)集進行復雜處理時CPU資源也是非常寶貴的.文獻[11]中將Spark任務中同一個Stage的RDD看作同一組,當緩存已滿或沒有足夠空間時,移除緩存中和待緩存RDD中不屬于同一組的RDD,如果都屬于同一組,則溢出到磁盤進行存儲,該方法在對RDD進行替換時僅考慮了一個Stage,這種粗粒度的緩存機制存在局限性.文獻[16]提出了AWRP(Adaptive Weight Ranking Policy)算法為每個緩存對象計算權重,并優(yōu)先轉換權重值最低的緩存對象,但權重值計算方法沒有對Spark的RDD緩存進行針對性的優(yōu)化,并不適用于分布式計算框架.

對于國內,文獻[12]提出了使用RDD權重值來進行緩存替換,將RDD的計算代價、使用次數(shù)、自身大小和生命周期作為衡量RDD權重的因素.然而作者并沒有對計算代價這一因素提出合理的計算方法,因此影響了RDD權重計算的準確性.文獻[9]提出了自適應緩存管理策略,同樣使用RDD權重值作為緩存替換的度量工具,借鑒了文獻[12]中影響權重值的四大因素,提出使用RDD生成時長來代替計算代價,并且忽略了和RDD大小有關的因素.這樣雖然使計算簡便了許多,然而并沒有解決RDD權重計算缺乏準確性這一缺陷,影響因素考慮不全面.作者在做RDD置換的時候沒有考慮到Spark系統(tǒng)在進行緩存置換的時候置換的單位是Block,對應的便是RDD的一個分區(qū).文獻[13]分析了Spark計算框架的內存管理和緩存機制,提出了基于RDD分區(qū)的權重計算,并以該權重值作為緩存置換的標準.但是文中僅僅考慮了RDD分區(qū)大小、RDD分區(qū)被使用次數(shù)和RDD分區(qū)存在內存中的時間,并沒有考慮影響一個RDD的其他因素,比如計算代價.這樣得到的權重值是不準確的,會影響Spark集群的計算效率.

總結起來,當前針對Spark緩存替換的研究工作還不夠完善,已有的基于權重的緩存替換算法存在權重值計算不準確,考慮因素不全面,度量方法不夠細致,影響了緩存的命中率和作業(yè)執(zhí)行的效率.本文結合現(xiàn)有大數(shù)據(jù)快速實時處理的需要,對Spark內核中RDD對象的緩存替換進行研究分析,提出了一種新的RDD分區(qū)權重緩存替換算法,包括權重計算和緩存替換的改進措施.通過細化分析RDD分區(qū)對象的影響因素,并對這些因素進行量化分析,使得 RDD分區(qū)權重值的計算更加精確.進而在內存資源不足的情況下,緩存中的RDD分區(qū)的替換更加合理.弱化了內存資源對整體性能的影響,讓集群效能得到了極大的發(fā)揮.

3 RDD分區(qū)權重新模型——WCSRP

Spark中最重要的抽象概念就是RDD,RDD通常是通過HDFS(或其他Hadoop支持的文件系統(tǒng))上的文件,或者驅動器中的Scala集合對象,來創(chuàng)建或轉換得到的.在Spark中一個RDD被劃分成若干個partition,因此對于RDDi可表示為RDDi={p1,p2,p3,…,pn}.用戶可以請求Spark將RDD緩存在內存中,當RDD被緩存在內存中,由于內存的容量是有限的,會出現(xiàn)內存存儲資源不夠用的情況.

由此,可以得出通過如下公式計算出RDD分區(qū)的權重:

w=α0CRDDi+α1FRDDi+α2Sp+α3LTRDDi+α4ILRDDi

(1)

公式(1)中 ，w表示該RDD分區(qū)的權重值，CRDDi表示該RDD的計算代價，F(xiàn)RDDi表示該RDD的使用次數(shù)，Sp表示該分區(qū)的大小，LTRDDi表示該RDD的生命周期，ILRDDi表示計算該RDD輸入RDD的位置。A={α0,α1,α2,α3,α4}中的元素是常數(shù)，分別是CRDDi、FRDDi、Sp、LTRDDi和ILRDDi的歸一化權重，權重值得選取由用戶的具體任務Task需求決定。

下面，我們將分別對CRDDi、FRDDi、Sp、LTRDDi和ILRDDi的權重進行量化計算。

3.1 RDD計算代價CRDDi

在Spark中RDD作為計算的參與者,在宏觀上,所有的計算過程都是根據(jù)RDD進行的,但從微觀上來看,算子的操作其實是作用在RDD的不同分區(qū)上.當一個RDD經過轉化操作派生出新的RDD時,Spark會使用譜系圖(lineage graph)來記錄這些不同RDD之間的依賴關系,Spark需要用這些信息來按需計算每個RDD,也可以依靠譜系圖通過用緩存中的RDD來恢復下游的RDD,而不需要從頭計算[17].因此RDD之間的依賴關系便成了整個應用運行的關鍵.在Spark源碼中Dependency類是依賴關系的基類,其中NarrowDependency和ShuffleDependency均繼承自該基類.前者表示RDD之間是窄依賴關系,這個的RDD會被劃到同一個Stage中,這樣就可以以管道的方式迭代執(zhí)行;后者表示RDD之間是寬依賴關系,依賴上游的RDD不止一個,且多個子分區(qū)會依賴于同一個父RDD分區(qū).

由于依賴關系的不同,在計算一個RDD的時候需要的進行計算的次數(shù)也是不同的.如圖1所示.

圖1 窄依賴和寬依賴結構圖Fig.1 Narrow dependency and wide dependency

其中,RDD A和RDD B屬于窄依賴,故RDD A的分區(qū)經過轉換操作一產生RDD B中的分區(qū).有圖可以看出RDD A和RDD B的分區(qū)數(shù)均是4,故計算產生RDD B時,需要在4個分區(qū)上進行計算,我們可以把這個計算的過程當做RDD B的計算代價,故RDD B的計算代價值為4.也就是說通過窄依賴得到的RDD的計算代價就是該RDD的父RDD分區(qū)數(shù).RDD C和RDD D屬于寬依賴,RDD C的每個分區(qū)都可能被RDD D所使用,RDD D的分區(qū)中的數(shù)據(jù)同樣來自RDD C所有的分區(qū),在進行計算的時候對RDD C的每個分區(qū)的操作次數(shù)等于RDD D的分區(qū)數(shù).從圖1可以知道RDD C的分區(qū)數(shù)為3,RDD D的分區(qū)數(shù)為3,那么總的計算次數(shù)便是9,和窄依賴相同我們也可以得到RDD D的計算代價,所以RDD D的計算代價值是9.這樣便得到了通過寬依賴得到的RDD的計算代價就是該RDD的分區(qū)數(shù)與父RDD的分區(qū)數(shù)的乘積.

由此,我們給出RDD計算代價的計算公式如下:

(2)

其中，parspRDDi表示屬于寬依賴的父RDD的分區(qū)數(shù)，a表示是否存在寬依賴，若不存在寬依賴a=0，若存在a=1，parnpRDDj表示屬于窄依賴的父RDD的分區(qū)數(shù)，b表示是否存在窄依賴，若不存在窄依賴b=0，若存在b=1.

3.2 RDD使用次數(shù)FRDDi

由于Spark的懶加載機制,使得Spark可以在Action算子觸發(fā)SparkContext.runJob之前,程序在運行時并沒有引入數(shù)據(jù)進行計算,此時系統(tǒng)通過分析代碼的邏輯,初始化通過Transformation算子產生的RDD(此時這些RDD中并沒有實際的數(shù)據(jù)).又因為每個RDD都有其依賴(除了最頂級RDD的依賴是空列表),所以可以確定各個RDD之間的依賴關系,然后根據(jù)RDD之間的依賴關系構建成DAG[18]圖,將不同的RDD串聯(lián)起來.在DAG圖中點對應RDD,邊對應一個算子.

下面,我們給出一個PageRank的例子,代碼如圖2所示.

圖2 PageRank的Spark實現(xiàn)代碼圖Fig.2 Implement PageRank in Spark

分析代碼可知,在調用saveAsTextFile這個Action算子之前,所有的Transformation算子都是為了構建RDD之間的依賴關系.而這些依賴關系也就構成了DAG圖,圖3便是PageRank在做兩輪迭代時的DAG圖.因為DAG圖中邊對應一個算子,那么也就是說這時該RDD被使用了一次,在圖3中我們可以看出links出去有三個箭頭,那么可以知道此時links被使用了三次.

所以我們可以通過統(tǒng)計DAG圖中某個RDD的出度,便可以得到該RDD的使用次數(shù).具體公式如下:

FRDDi=N

(3)

其中,N表示RDDi的出度.

3.3 RDD分區(qū)大小Sp

在Spark中使用BlockManager來緩存RDD數(shù)據(jù),程序內部定義了抽象類BlockStore,用于制定所有存儲類型的規(guī)范.目前BlockStore的具體實現(xiàn)包括MemoryStore、DiskStore和TachyonStore.其中MemoryStore是負責將沒有序列化的Java對象或序列化的ByteBuffer存儲到內存中.開發(fā)者使用 persist()或者 cache()函數(shù)來標記一個 RDD 是持久化的,當這個 RDD 在被一個action觸發(fā)的作業(yè)提交計算后,它就會緩存在內存中.我們都知道,RDD 的運算是基于partition,每個task代表一個分區(qū)上一個stage內的運算閉包,task被分別調度到多個executor上去運行,運算過程中的RDD若需要存儲則會將在該executor上運行的partition緩存下來對應的就是Block,Spark中對存儲內容的讀取就是根據(jù)Block進行的.本文討論的是在內存中的緩存,故在緩存的時候會調用到MemoryStore中的實現(xiàn)方法putIteratorAsValues或putIteratorAsBytes,這兩個方法的區(qū)別在于一個是負責嘗試將沒有序列化的Java對象放入內存,另一個是試著將序列化的ByteBuffer放入內存,這兩個均是嘗試待緩存的數(shù)據(jù)大小是否超過當前剩余的空閑內存,如果未超過則進行存儲操作,完成緩存動作.

圖3 PageRank的DAG圖Fig.3 DAG graph of PageRank

因此可以通過監(jiān)聽,獲得準備緩存的分區(qū)的大小,公式如下:

Sp=sacquireMemory

(4)

其中,sacquireMemory表示該分區(qū)申請內存的大小.

3.4 RDD生命周期LTRDDi

RDD的生命周期(LTRDDi)就是該RDD存活的時間段,如果一個RDD不會再被使用我們可以看做該RDD已經死亡,反之一個RDD還可以通過轉換操作產生新的RDD那么該RDD還處于存活狀態(tài).對于生命周期的計算可以通過DAG圖結合應用的執(zhí)行過程進行分析.Spark在進行任務處理的時候,計算框架內管理線程級別的Task,在進行任務調度的時候,一個Stage中的Task會被分別調度到計算節(jié)點.Task在運行的時候可以看做pipeline.一個RDD由多個partition組成,每個partition經過Transformation操作產生新的RDD中對應的partition,依次進行下去直到該Stage結束.這樣的過程便是一個Task的運算過程,當Task在一個Executor上進行運算時,它是串行執(zhí)行的.故某個RDD的生命周期是該RDD的第一個partition產生后一直到最后一個使用該RDD所在的Stage完成計算的這段時間.如圖4所示.

圖4 Stage中Task的執(zhí)行過程圖Fig.4 Execution of the Task in the Stage

故對于RDDi的生命周期就是RDD生成第一個分區(qū)的時間fpTRDDi與最后一次使用該RDD的Stage完成計算時間eTstagei之差.我們給出的公式如下:

LTRDDi=eTstagei-fpTRDDi

(5)

3.5 RDD計算位置ILRDDi

在Spark中任務的處理也要考慮數(shù)據(jù)的本地性,Spark目前支持PROCESS_LOCAL(本地進程)、NODE_LOCAL(本地節(jié)點)、NO_PREE(沒有喜好)、PACK_LOCAL(本地機架)、ANY(任何).Spark的設置中有關于本地進程、本地節(jié)點和本地機架等的等待時間.當某一個Task在根據(jù)自己的locality level執(zhí)行時,由于啟動失敗,然后以自己當前的locality level等待第二次被調度,若等待時間超過了該本地化級別的默認等待時間,則該Task會通過降低自己的locality level來嘗試被再次啟動.我們知道,對于很多Task來說,執(zhí)行時間往往比網絡傳輸和磁盤I/O的耗時要短得多.本文將不同的Locality Level進行量化依次是:1、2、3、4、5,隨著數(shù)字的增大代表著Task計算節(jié)點與Task的輸入數(shù)據(jù)的節(jié)點距離越來越遠,這樣也就產生了網絡傳輸對Task執(zhí)行時間的影響,進而也影響著RDD的計算效率.通過在Spark源碼中內嵌代碼,統(tǒng)計RDDi的各個partition對應的Task運行的Locality Level,然后進行疊加計算出RDDi的輸入RDD位置值.由于在Task運行的過程中,是在一個executor上進行的串行式的計算,所以除Task起始的那個RDD其他RDD的輸入位置都是本地進程.

由此,我們給出公式如下:

(6)

其中,pLk表示RDDi中的第k個分區(qū)的輸入節(jié)點的Locality Level.

4 基于WCSRP模型的緩存替換算法

由公式(1)-公式(6)可得:

(7)

根據(jù)公式(7)我們提出基于WCSRP模型的緩存替換算法.具體操作如算法1所示.

算法1. 基于WCSRP模型的緩存替換算法

輸入： RDD分區(qū)權重集合 wList

空閑緩存大小 freememory

待緩存RDD分區(qū)的大小 size

待緩存RDD分區(qū)的權重 weight

輸出：緩存成功 true; 緩存失敗 false

初始化：

//待替換列表

rptlist<-new List

//待替換列表的總大小

rptlistsize = 0

//替換列表

rplist<-new List

for i=0 to wList.length-1 do

if weight > wList[i].weight then

rptlist.add(wList[i]);

rptlistsize+=wList[i].size;

end if

end for

if rptlist.Length==0 then

return false;

end if

if rptlistsize+freememory

rptlist.clean();

return false;

else

rptlist.orderByWeight(); // 重排序

for i=0 to rptlist.Length-1 do

rplist.add(rptlist[i]);

freememory+=rptlist[i].size;

if freememory>size then

for j=0 to rplist.Length-1 do

//移除替換列表中的Block

delete(rplist[j]);

end for

rplist.clean();

return true;

end if

end for

rptlist.clean();

return false;

end if

算法1的具體實施過程如下:

1.首先,獲取需要緩存的RDD分區(qū)的大小和該RDD分區(qū)的權重值.

2.然后對緩存中的Block進行過濾,將權重值小于待緩存RDD分區(qū)權重值的加入待替換列表.

3.若待替換列表為空,則停止替換,不緩存該RDD的分區(qū).若不為空,判斷待替換列表的大小和空閑空間的和是否小于申請大小,若小于則停止替換并清空待替換列表,若大于則將列表按權重值從小到大的順序排列.

4.遍歷待替換列表中的Block,依次加入替換列表,直到空閑內存與替換列表中Block總大小的和大于等于待緩存RDD分區(qū)的大小,則停止遍歷,將替換列表中的Block替換出緩存,將待緩存RDD分區(qū)加入緩存,并清空替換列表和待替換列表.

5.若待替換列表中的Block的總大小與空閑內存空間的大小的和小于待緩存RDD分區(qū)的大小,則停止替換,不緩存該RDD的分區(qū),清空待替換列表.

5 理論分析與實驗評價

5.1 理論比較分析

當前Spark權重緩存替換算法,通常涉及RDD計算代價、RDD使用頻率、RDD分區(qū)大小和RDD生命周期等關鍵參數(shù).下面我們將本文提出的WCSRP與文獻[9]提出SACM、文獻[13]提出的DWRP以及Spark平臺現(xiàn)有的置換算法LRU進行比較.如表1所示.

表1 相關緩存置換算法對比Table 1 Comparison of correlation cache replacement algorithms

對于SACM,計算權重時考慮了RDD使用頻率和RDD計算代價,提供參數(shù)校準.和本文提出的WCSRP相比,考慮的因素不夠全面,另外其在進行內存置換時的置換目標是RDD,當出現(xiàn)內存置換時,會影響其他Task的執(zhí)行.因此,在采用本文WCSRP替換算法,不會出現(xiàn)Task運行時所使用的RDD分區(qū)被替換出去需要重新計算的情況,應用的運行時間會比SACM有所降低.當運行長作業(yè)時,降低的幅度尤為明顯.

對于DWRP,計算權重時考慮的因素有RDD使用頻率、RDD分區(qū)大小和RDD生命周期.和WCSRP相比較,其在進行權重計算時僅考慮了三個因素,而RDD計算代價和Locality Level這兩個因素對應用的執(zhí)行時間有著很大影響,另外權重計算公式中沒有提供參數(shù)校準.所以最終計算得到的權重值沒有本文提供的公式計算的權重值精確,在進行內存置換時對Spark作業(yè)執(zhí)行效率的提升沒有本文WCSRP明顯.

最后,對于Spark自帶的LRU置換算法,在進行內存置換時忽略了RDD分區(qū)的差異性,僅考慮當前內存中的RDD分區(qū)的訪問順序,而當高重用但最近未使用或重新計算耗費極高代價的RDD的分區(qū)被替換出去,應用執(zhí)行的耗時增加,整個框架作業(yè)運行效率會降低.而本文的WCSRP替換算法綜合考慮了影響RDD分區(qū)的多個因素,使得在進行替換時不是盲目的僅根據(jù)當前的訪問順序.這樣可以省去不必要的重新計算所花費的時間,提高應用的運行效率.

5.2 實驗評價

為了實驗驗證WCSRP,我們在Spark平臺上實現(xiàn)了WCSRP,WCSRP各因素初始的權重值是A={а0,а1,а2,а3,а4}={0.3,0.2,0.2,0.2,0.1}.為了與本文提出的WCSRP形成對照,我們選取Spark自帶的緩存替換算法LRU來作對比.

實驗在一臺服務器DELL PowerEdge T620(Intel Xeon E5-2620*2/32G/1T)上進行,操作系統(tǒng)使用Debian 8.7,在服務器上虛擬出了4臺主機,利用這四臺主機搭建了Spark集群.其中一個主機作為Spark的Master節(jié)點,其他三個作為worker節(jié)點.應用提交后的運行時間可以通過Spark的日志文件得到.實驗數(shù)據(jù)選擇由SNAP提供的標準數(shù)據(jù)集Amazon0601,實驗是測試使用不同的緩存替換機制,在不同迭代次數(shù)下PageRank算法的執(zhí)行時間.選擇PageRank算法來進行實驗的原因是因為PageRank算法是典型的數(shù)據(jù)密集型算法,會涉及到多次迭代,當使用緩存后會有效的提升計算的效率.

首先測試PageRank算法使用Spark默認的緩存替換算法,記錄不同的迭代次數(shù)進行實驗的結果,每個迭代次數(shù)進行5次實驗,分別記錄下執(zhí)行的時間,然后得出平均值便是該迭代次數(shù)的執(zhí)行時間.然后使用新的權重緩存替換算法進行同樣的實驗,對記錄的數(shù)據(jù)求平均值.實驗結果如表2所示.

表2 LRU和WCSRP對比試驗的統(tǒng)計結果Table 2 Result of Comparative test between LRU and WCSRP 單位：秒(S)

通過上表,便可得到圖5.

圖5 PageRank使用不同緩存替換算法對比試驗Fig.5 PageRank uses different cache replacement algorithms to experiment

實驗結果顯示,在相同的迭代次數(shù)使用不同的緩存替換算法的情況下,PageRank在Spark框架中的執(zhí)行時間是不同的.盡管WCSRP和LRU均隨著迭代次數(shù)的增加,執(zhí)行時間也在增加.但是使用WCSRP算法作為緩存替換算法時應用的執(zhí)行時間明顯比使用LRU算法作為緩存替換算法時有所降低.因為在使用LRU算法時,僅僅只是考慮內存中的Block被訪問的時間,長期未被訪問的便會被置換出去,它忽略了該Block對整個應用的價值.而在使用WCSRP算法時我們不僅綜合考慮RDD的計算代價、使用次數(shù)、分區(qū)的大小和生命周期四大因素對權重的影響,而且還增加考慮了Task執(zhí)行時locality level這個因素,準確計算出了緩存中的每個Block的權重,將對該應用執(zhí)行最有價值的Block繼續(xù)緩存在內存中,防止在需要的時候進行重復計算,使得有限的內存資源得到了充分的利用,減少了重復計算的出現(xiàn),也便提高了整個應用執(zhí)行的效率.所以才出現(xiàn)了實驗結果中的執(zhí)行時間減少的現(xiàn)象.

6 總 結

本文提出了新的權重緩存替換算法,綜合考慮了影響RDD緩存的各大因素,改進了之前提出的RDD權重緩存替換策略,用細粒度的RDD分區(qū)權重值來作為替換的度量標準,改進了各影響因素的量化方式.通過這種方法,提高了Spark框架應用執(zhí)行的效率,減少了因內存異常導致任務執(zhí)行失敗的情況出現(xiàn).理論和實驗結果均證明了新的權重緩存替換算法對Spark框架有著明顯的優(yōu)化作用.