陳天宇，張龍信，李肯立，周立前

1(湖南工業(yè)大學(xué) 計算機學(xué)院，湖南 株洲 412007)2(湖南大學(xué) 信息科學(xué)與工程學(xué)院，長沙 410082)

1 引 言

隨著大數(shù)據(jù)與云計算的興起,基于集群的大規(guī)模數(shù)據(jù)處理已成為各大IT公司的解決方案[1].許多高科技企業(yè)使用的大數(shù)據(jù)分析技術(shù)具有迭代特性,包括使用圖計算來進行PageRank或社交網(wǎng)絡(luò)分析,使用機器學(xué)習(xí)進行聚類或回歸分析等.這些應(yīng)用共同的特點是其數(shù)據(jù)需要多次迭代處理直到滿足收斂條件或者結(jié)束條件,于此同時,大量的數(shù)據(jù)需要在迭代中重用[2].

目前比較有影響力的大數(shù)據(jù)處理框架有Hadoop、Spark、Storm等[3],它們均源于Google早期提出的MapReduce思想.Hadoop在進行迭代計算時速度比較慢,當待處理的數(shù)據(jù)規(guī)模越來越大時,Hadoop通過分布式文件系統(tǒng)讀寫的性能瓶頸愈發(fā)明顯,而Spark正是在這種背景下應(yīng)運而生.Spark針對迭代計算中需要被多次使用的工作數(shù)據(jù)集進行優(yōu)化,引入了內(nèi)存集群計算的概念,將數(shù)據(jù)集緩存在內(nèi)存中,以減少訪問延遲.用戶可以調(diào)整Spark的存儲策略,Spark是基于彈性分布數(shù)據(jù)集(RDD)的內(nèi)存計算,數(shù)據(jù)集可從記錄的信息來源重構(gòu).RDD被表示為一個對象,并且可以在文件中創(chuàng)建[4].

由于Spark的存儲策略可由用戶選擇定制,Spark整個計算過程有不同的存儲策略可供選擇[5].用戶可以根據(jù)經(jīng)驗選擇緩存到內(nèi)存或是物理硬盤中,由此產(chǎn)生的計算時間不盡相同.當緩存的是無用數(shù)據(jù)時,該數(shù)據(jù)會占用內(nèi)存空間,系統(tǒng)性能將降低.如果緩存了錯誤的數(shù)據(jù)還會導(dǎo)致內(nèi)存溢出等后果,甚至可能將隨后重用的中間結(jié)果忽略,造成計算時間延長.通常情況下,計算機內(nèi)存空間是有限的,無法保證每個計算節(jié)點有充裕的內(nèi)存空間緩存數(shù)據(jù).

目前的存儲替換策略中,先入先出(FIFO)策略主要關(guān)注創(chuàng)建時間,最近最少使用(LRU)策略更側(cè)重于命中歷史訪問次數(shù),然而這些經(jīng)典的算法沒有考慮分片的計算成本問題[6].現(xiàn)有的研究大多數(shù)基于此類算法改進,其中部分改進算法諸如GreedyDual[7]、GD-Wheel[8]均考慮了訪問歷史和計算成本,但在一個作業(yè)中待處理任務(wù)的執(zhí)行邏輯在Spark中是已知的,對訪問RDD歷史次數(shù)進行優(yōu)化的緩存策略效果不明顯.類似的研究還有不考慮RDD的訪問歷史次數(shù)的LCS[9]算法,以及考慮計算成本次數(shù)加權(quán)的WR[10]算法.本文針對內(nèi)存占用率與RDD替換時的權(quán)重,考慮了分片的歷史訪問次數(shù)和計算成本等因素對權(quán)值的影響,并提出緩存權(quán)重替換算法(CWS).

本文第一部分介紹了Spark的應(yīng)用背景;第二部分闡述了Spark的相關(guān)概念;第三部分介紹相關(guān)的預(yù)備知識;第四部分詳細介紹本文提出的算法;第五部分通過實驗驗證了本文提出的算法的性能;最后對本文工作進行了總結(jié).

2 Spark相關(guān)概念

2.1 彈性分布式數(shù)據(jù)集

彈性分布式數(shù)據(jù)集的提出,是對不適合非循環(huán)數(shù)據(jù)流模型應(yīng)用的一種改進.非循環(huán)數(shù)據(jù)流模型是從物理存儲中加載記錄,將操作記錄傳入有向無環(huán)圖(DAG)后寫回物理存儲介質(zhì)中[11].計算機集群系統(tǒng)通過調(diào)用這個數(shù)據(jù)流圖來完成調(diào)度工作和故障恢復(fù).當作業(yè)在多個并行操作時需要重用工作數(shù)據(jù)集(例如迭代算法、交互式數(shù)據(jù)挖掘[12]),系統(tǒng)會將數(shù)據(jù)輸出到物理存儲介質(zhì),計算機集群每次重用時都需要重新加載,從而導(dǎo)致開銷較大.

RDD是Spark框架的核心抽象,作為Spark的用戶,可以把RDD看成獨立數(shù)據(jù)分片的集合.RDD具有以下特征：

1) RDD能夠記錄其血統(tǒng)(Lineage)信息,RDD能夠記錄其本身是怎樣通過其他數(shù)據(jù)集來產(chǎn)生或者轉(zhuǎn)換的.在某個RDD分片丟失時,能夠根據(jù)Lineage信息恢復(fù)該分片,而不必重新計算所有分片.

2) RDD中的數(shù)據(jù)分片存儲在集群中的節(jié)點上,可并行操作每個分片中的數(shù)據(jù).RDD的每個分片上都有函數(shù),通過函數(shù)調(diào)用可操作RDD的分片轉(zhuǎn)換.

3) Spark在任務(wù)調(diào)度時,盡可能將計算任務(wù)分配到數(shù)據(jù)塊所存儲位置,即數(shù)據(jù)本地化.

RDD自身是只讀數(shù)據(jù)集,僅能通過對其它的RDD執(zhí)行轉(zhuǎn)換操作(例如map,join和groupBy)創(chuàng)建.傳統(tǒng)的分布式共享內(nèi)存尋找丟失的分片需要checkpoint和rollback操作,開銷較大,而RDD的內(nèi)存共享方式既能保證低開銷,又能保證容錯性.RDD通過轉(zhuǎn)換操作轉(zhuǎn)換成不同的RDD后,可通過lineage來計算出相應(yīng)的RDD分片.RDD所有分片都分布在集群系統(tǒng)的各節(jié)點上,默認情況下的RDD在需要用到時都會被重新計算.

2.2 Spark中的作業(yè)調(diào)度與緩存機制

在Spark中,RDD被表示為對象,RDD的操作僅包含轉(zhuǎn)換操作和行為操作.RDD在轉(zhuǎn)換操作執(zhí)行完后才可能執(zhí)行行為操作,此時DAGScheduler為每個執(zhí)行中的任務(wù)生成一個DAG(有向無環(huán)圖),隨后將任務(wù)上傳至集群.圖1是Spark中的作業(yè)調(diào)度模型,本文對于選擇算法部分的優(yōu)化基于DAG將選擇RDD分片的緩存至內(nèi)存.

圖1 Spark中的任務(wù)調(diào)度Fig.1 Task scheduling in the Spark

圖2展示了RDD在集群中的緩存,RDD2是由RDD1經(jīng)過Transformation操作轉(zhuǎn)換得到的,RDD2在接下來的運行過程中有可能被再次使用,所以Spark主動緩存RDD2.假設(shè)RDD2中包含3個分片21、P22和P23,集群中有3個節(jié)點,緩存時P21、P22和P23分別存儲在節(jié)點1、節(jié)點2以及節(jié)點3的內(nèi)存中.

圖2 集群中的RDD緩存Fig.2 RDD cache in a cluster

在Spark默認的緩存機制中,當RDD在內(nèi)存中完成計算后,可通過CacheManager來獲取結(jié)果.RDD分片由CacheManager來緩存,而CacheManager中的操作取決于BlockManager提供的API.用戶通過BlockManager來決定分片將從內(nèi)存或者磁盤中獲取,而MemoryStore則確定分片是否緩存至內(nèi)存中.當分片占滿內(nèi)存之后,默認情況下系統(tǒng)將用LRU算法來選擇要被替換的分片[13].傳統(tǒng)的LRU算法忽略了RDD分片的大小和計算成本,無法確定所選擇的分片是否將被重用.本文的工作正是針對RDD分片的選擇和替換策略進行優(yōu)化.

3 預(yù)備知識

3.1 問題定義

Spark在任務(wù)執(zhí)行過程中需要用到某個RDD時,并非立即對該RDD進行計算,任務(wù)調(diào)度器會根據(jù)該RDD的lineage關(guān)系構(gòu)建一個DAG圖.生成DAG后,Spark會將任務(wù)劃分為不同的stage,執(zhí)行后得到目標的RDD.在本文中用表示分片,Pij表示第i個RDD的第j個分片.

當剩余的緩存不足以存放新的RDD時,Spark默認的緩存策略會替換最近最少使用的RDD.在一個作業(yè)中RDDn所包含的分片Pnm被推出緩存后,若在后續(xù)的作業(yè)中被重用,該分片會被再次計算,采用默認的算法會產(chǎn)生很多不必要的開銷,我們需要解決默認情況下僅考慮最近最少使用的逐出帶來的計算效率差異.

在計算RDD的過程中,任務(wù)的stage通過DAG來劃分,本文提出的替換策略在執(zhí)行過程中將stage中滿足替換條件的分片進行替換.

圖3 一個計算過程的DAG圖Fig.3 DAG graph of a computational process

圖3展示了Spark中一個作業(yè)執(zhí)行階段的DAG圖,假設(shè)該作業(yè)的所有分片都在同一個節(jié)點之中,實驗通過六臺虛擬機所組成的節(jié)點計算數(shù)據(jù).這個任務(wù)序列的執(zhí)行階段Stage如公式(1)和公式(2)所示.

Stage11{RDD3→RDD4→RDD5}

(1)

Stage21{RDD3→RDD5}

(2)

RDD是由分片組成的,假定一個作業(yè)中所有的RDD分片大小都相同,用Pnm表示第n個RDD的第m個分片,那么:

RDDn=Pn1+…+Pnm+…+Pnp

(3)

假設(shè)每個分片的執(zhí)行時間為TPnm,則Stage2中的執(zhí)行總時間為：

TP=TP32+2TP31+TP41+TP42

(4)

我們可以看到P31被重復(fù)計算了,而這僅僅是一個作業(yè)中的小部分,整個作業(yè)中可能存在大量重復(fù)的高開銷計算.

3.2 計算模型

Spark在數(shù)據(jù)處理的過程中,為每個作業(yè)生成一個DAG.在作業(yè)的計算過程中,有些變量會反復(fù)出現(xiàn),若將這些變量緩存在內(nèi)存中可以顯著提升處理速度.Spark默認內(nèi)置的LRU是通過LinkedHashMap實現(xiàn)的,本文所實現(xiàn)的工作在此基礎(chǔ)上重構(gòu)了內(nèi)存管理策略.

默認的LRU算法選擇最近使用的RDD分片,不考慮分片的計算成本和大小.當兩個RDD分片的大小和重復(fù)出現(xiàn)的次數(shù)相同時,應(yīng)該緩存計算成本更高的分片;而當兩個分片的計算成本和重復(fù)出現(xiàn)的次數(shù)都相同時,則需要考慮緩存空間較小的分片.

4 緩存權(quán)重替換策略(CWS)

4.1 RDD計算過程

通常情況下,當計算緩存資源不夠用時,需要對緩存中的RDD分片進行替換,我們用MemoryC表示集群內(nèi)存總資源,計算節(jié)點中實際空閑內(nèi)存記為MemoryA,默認條件下,假設(shè)此時的任務(wù)中有q個RDD在等待,則等待計算的RDD所需占用的內(nèi)存資源必須小于MemoryA.即：

(5)

Spark中的行為操作都是在轉(zhuǎn)換操作之后才執(zhí)行的,所以可從DAG圖中統(tǒng)計分片重復(fù)使用的次數(shù),在作業(yè)執(zhí)行的過程中,本文把緩存中已有的RDD記為MemoryS.

4.2 RDD的選擇

默認情況下系統(tǒng)使用LRU算法淘汰最長時間未被使用的分片,如果該分片需要被再次使用,此分片重新計算的成本可能會非常高.LRU無法確定分片是否有保存的意義,本文提出的CWS算法能選擇高權(quán)值的分片并保留在緩存中,該算法的偽代碼如算法1所示.

算法1.選擇算法

Selection algorithm

actual free cacheMemoryA

maximum cacheMemoryC

selected blocksMemoryS

Output:evicted RDDs

1 get a DAG from the driver of the spark

2 allocate memory according to DAG

3if(MemoryE

4fori=1toq

5if(MemoryE

6MemoryA

7elseif(MemoryE>MemoryA)

8 call replacement algorithm

9endif

10endfor

11endif

在選擇算法中,MemoryE表示等待執(zhí)行隊列中的RDD,RDD的數(shù)量記為q個,剩余的緩存總量表示為MemoryC.算法首先在設(shè)定好緩存的替換策略后,遍歷內(nèi)存查找是否已有待選擇的RDD,若已有則停止循環(huán).第5-9步判斷實際空閑內(nèi)存的大小,當實際空閑內(nèi)存大于待分配的內(nèi)存大小時,將RDD放入緩存中,否則跳轉(zhuǎn)至替換策略.該選擇算法的時間復(fù)雜度為O(n).

4.3 權(quán)值替換

Spark中的各節(jié)點處理能力基本相同,那么分片完成作業(yè)花費的時間更多,則意味著其計算成本更高.通常情況下,我們可以使用分片大小近似作為其計算成本.在替換分片的過程中,當發(fā)現(xiàn)某個分片占用的存儲空間較大(計算成本更高)而使用次數(shù)又較少時,本文提出的算法將用存儲空間較小且使用次數(shù)相同的分片替換該分片,或者使用存儲空間大小相近且使用次數(shù)更多的分片來替換該分片.在公式(6)中,當分片的使用次數(shù)Fnm不變時,分片的計算成本Spnm的增加將導(dǎo)致該分片的權(quán)值變小;當Spnm不變時,該分片的權(quán)值隨著Fnm增加而變大.本文用Fnm表示RDDn的第m個分片的使用次數(shù),分片大小記為Spnm,分片的權(quán)值Vpnm則可表示為：

Vpnm=Fnm/Spnm

(6)

若RDDn有m個分片,那么RDDn的權(quán)值VRn為：

(7)

當內(nèi)存中的分片緩存接近飽和時,替換算法將根據(jù)Vpnm和VRn替換權(quán)值較小的分片.替換算法的偽代碼如算法2所示.

算法2.替換算法

Replacement algorithm

Input：selected blocksMemoryS

the valueVRnof candidate RDDs

the size of partitionsSpnm

execution sequencePnm

Output：evicted RDDs

1fori=1toq

2if(Pnm==Pni)

3 break;

4elsequickly sortMemorySforVRn

5 expel RDD;

6endif

7endfor

在替換算法中,首先判斷待替換分片是否存在于內(nèi)存中,若不存在則對內(nèi)存中待替換的緩存序列按權(quán)值進行快速排序,得到新的序列,最后按照新的序列逐出RDD.該算法的時間復(fù)雜度為O(n).

5 實驗部分

5.1 實驗環(huán)境

為避免因?qū)嶒灜h(huán)境差異帶來的實驗結(jié)果對比的不便,本文盡量使用與目前主流研究算法一致的實驗環(huán)境.在服務(wù)器上部署含有6個節(jié)點的集群,每個節(jié)點配備8核2.2GHz Intel Xeon E5-2620 CPU、物理硬盤空間80G、內(nèi)存根據(jù)實驗條件可調(diào)整為2G、4G、8G等多種情況.集群使用的操作系統(tǒng)為CentOS 7,Scala版本為2.11.8,而Java開發(fā)工具包版本為1.7,Hadoop的版本為2.2.0,Spark版本為2.0.0.實驗過程中的計算時間通過對三次運行結(jié)果取平均值,內(nèi)存占用率則通過Ganglia[14]監(jiān)控獲取.

測試數(shù)據(jù)集使用斯坦福大學(xué)提供的公開網(wǎng)絡(luò)分析項目獲取的17個真實的圖形數(shù)據(jù)集.這些數(shù)據(jù)集的Nodes和Edges對執(zhí)行時間和內(nèi)存的使用情況影響較大.實驗采取PageRank算法對這些數(shù)據(jù)集進行排序,數(shù)據(jù)集如表1所示.

5.2 內(nèi)存充足條件下的算法對比

表1 斯坦福大型網(wǎng)絡(luò)數(shù)據(jù)集
Table 1 Stanford large network dataset

NameNodesEdges Descriptionp2p-Gnutella041087639994Gnutella peer to peer network from August 4,2002p2p-Gnutella242651865369Gnutella peer to peer network from August 24,2002wiki-Vote7115103689Wikipedia who-votes-on-whom networkp2p-Gnutella3162586147892Gnutella peer to peer network from August 31,2002Cit-HepTh27770352807Arxiv High Energy Physics paper citation networksoc-sign-Slashdot08110677357516757Slashdot Zoo signed social network from November 6,2008Cit-HepPh34546421578Arxiv High Energy Physics paper citation networksoc-sign-Slashdot09022182144549202Slashdot Zoo signed social network from February 21,2009Soc-sign-epinions131828841372Epinions signed social networkSlashdot090282168948464Slashdot social network from November 2008Amazon03022621111234877Amazon product co-purchasing network from March 2,2003Web-Stanford2819032312497Web graph of Stanford.eduAmazon03124007273200440Amazon product co-purchasing network from March 12,2003Wiki-Talk23943855021410Wikipedia talk (communication) networkweb-Google8757135105039Web graph from Googlecit-Patents377476816518948Citation network among US patentssoc-Pokec163280330622564Pokec online social network

由于Spark是基于內(nèi)存計算的框架,所以集群系統(tǒng)中各節(jié)點的內(nèi)存空間越充足其計算優(yōu)勢越明顯.當緩存重復(fù)的RDD分片時,充足的內(nèi)存能保證最大限度地縮短執(zhí)行時間.在本文實驗中,根據(jù)目前國內(nèi)外研究均使用的PageRank算法來測試性能,使用開源項目Ganglia監(jiān)控內(nèi)存使用率[15].每個數(shù)據(jù)集在Spark集群環(huán)境下運行PageRank算法三次,取其時間的平均值.本文使用表1中的數(shù)據(jù)集來測試充足內(nèi)存條件下的執(zhí)行時間和內(nèi)存占用率,分別測試了默認情況的下LRU、權(quán)重替換WR和CWS的策略.

圖4(a)是每個數(shù)據(jù)集在內(nèi)存空間寬裕的條件下對應(yīng)的執(zhí)行時間.當每個數(shù)據(jù)集在充裕的內(nèi)存中被處理時,WR的替換算法優(yōu)勢并不明顯,而WR的選擇算法由于執(zhí)行選擇算法時會頻繁統(tǒng)計分片的使用次數(shù),數(shù)據(jù)集較小時,其處理時間可能會稍微增加.但是,隨著數(shù)據(jù)集的Nodes和Edges數(shù)量的增加,WR算法的計算時間會減少.CWS算法在計算過程中減少了統(tǒng)計分片的使用次數(shù),降低內(nèi)存占用率的同時允許計算時間適當?shù)卦黾?計算較小的數(shù)據(jù)集時,由于減少了頻繁的統(tǒng)計分片使用次數(shù),從而縮短了計算時間.計算較大的數(shù)據(jù)集時,算法側(cè)重于降低內(nèi)存的占用率,執(zhí)行時間不是算法重點考慮的指標.CWS算法在處理表1所示的圖形數(shù)據(jù)集時,總的執(zhí)行時間比WR算法平均降低了2.4%.在圖4(b)中,當每個節(jié)點使用8G內(nèi)存時,WR算法的內(nèi)存占用率明顯高于其它算法,CWS盡管在計算時間上沒有明顯的優(yōu)勢,但其內(nèi)存的占用率較低.

5.3 有限內(nèi)存條件下的算法對比

Spark在內(nèi)存接近飽和時使用LRU算法來重新分配內(nèi)存,在有限的內(nèi)存情況下替換算法起著重要作用.我們將LRU、WR和CWS算法分別在單個節(jié)點內(nèi)存大小分別為2G和4G條件下進行比較.我們使用表1的數(shù)據(jù)集運算PageRank算法測試,使用Ganglia監(jiān)控內(nèi)存使用率.每個數(shù)據(jù)集進行三次運算后取平均值,發(fā)現(xiàn)每個節(jié)點的內(nèi)存使用情況大致相同.因為Spark使用負載均衡策略來管理資源,所以我們用其中一個節(jié)點的內(nèi)存使用情況對比.圖5(a)展示了單個節(jié)點在2G內(nèi)存條件下完成較小數(shù)據(jù)集的處理的執(zhí)行時間,圖6(a)展示了單個節(jié)點在4G內(nèi)存條件下完成較大數(shù)據(jù)集的執(zhí)行時間.

圖4 單節(jié)點使用8G內(nèi)存時數(shù)據(jù)集運行不同算法后的執(zhí)行時間和內(nèi)存占用率Fig.4 Task execution time and memory usage comparison under different algorithms with RAM=8 GB in each node

圖5 單節(jié)點使用2G內(nèi)存時數(shù)據(jù)集運行不同算法后的執(zhí)行時間和內(nèi)存占用率Fig.5 Task execution time and memory usage comparison under different algorithms with RAM=2 GB in each node

從圖5(a)和圖6(a)的對比,我們可以看出,在內(nèi)存有限的情況下,計算量較大的數(shù)據(jù)集會導(dǎo)致緩存空間不夠,LRU算法僅根據(jù)分片最近是否使用來決定是否替換,當被替換的分片計算成本較高時,此時可能需要花費更多時間;WR算法會頻繁計算分片的使用次數(shù),當數(shù)據(jù)集較小時,也會消耗掉部分時間.從圖5(b)不難看出,2G內(nèi)存條件下各算法的內(nèi)存占用率基本一致,與LRU相比,WR算法的平均計算時間有較大改善,CWS算法在計算時間節(jié)省方面沒有較大提高.在圖6(b)中,每個節(jié)點使用4G內(nèi)存時,CWS算法相比WR算法降低了0.8%的內(nèi)存占用率,而WR算法的內(nèi)存占用率會隨著數(shù)據(jù)集的增大高于其它算法.

圖6 單節(jié)點使用4G內(nèi)存時數(shù)據(jù)集運行不同算法后的執(zhí)行時間和內(nèi)存占用率Fig.6 Task execution time and memory usage comparison under different algorithms with RAM=4 GB in each node

6 總 結(jié)

通過對Spark的RDD進行深入研究,我們從選擇RDD和替換RDD兩個方面對Spark的緩存機制進行算法調(diào)優(yōu).Spark是基于內(nèi)存的計算,但Spark默認的機制沒有充分利用內(nèi)存的性能.WR算法雖然考慮了多個因素,但由于頻繁的選擇操作,在處理較小數(shù)據(jù)集時,產(chǎn)生了較多的時間開銷.本文在此基礎(chǔ)上,對算法進行優(yōu)化.實驗結(jié)果表明CWS算法在內(nèi)存空間充裕的條件下處理較小數(shù)據(jù)的平均執(zhí)行時間相比WR算法縮短了2.4%,內(nèi)存占用率降低36%;在內(nèi)存空間有限的條件下,每個計算節(jié)點使用2G內(nèi)存時,內(nèi)存占用率與WR算法基本持平.使用4G內(nèi)存時,與WR算法相比,CWS算法降低了0.8%的內(nèi)存占用率.