(重慶市北碚區(qū)中醫(yī)院 重慶 400700)

Spark計(jì)算節(jié)點(diǎn)同構(gòu)環(huán)境下Executor的內(nèi)存分配優(yōu)化模型

朱蓉

(重慶市北碚區(qū)中醫(yī)院重慶400700)

在對(duì)Spark云計(jì)算平臺(tái)的作業(yè)執(zhí)行機(jī)制進(jìn)行研究分析的基礎(chǔ)上，針對(duì)目前Spark處理作業(yè)時(shí)Executor的人為請(qǐng)求資源機(jī)制，提出了一種Spark計(jì)算節(jié)點(diǎn)同構(gòu)環(huán)境下Executor所需內(nèi)存資源的優(yōu)化分配算法，該算法能夠基于處理數(shù)據(jù)的數(shù)據(jù)量對(duì)分布式內(nèi)存資源進(jìn)行彈性分配，達(dá)到了Spark在數(shù)據(jù)量變化情況下合理利用分布式內(nèi)存資源這一目的，并對(duì)此進(jìn)行了仿真實(shí)驗(yàn)驗(yàn)證。

Spark；Executor；內(nèi)存資源；資源分配

一、前言

近幾年來(lái)，云計(jì)算技術(shù)成為人們討論的熱點(diǎn)，隨著云計(jì)算研究的不斷深入和發(fā)展，Spark大數(shù)據(jù)處理平臺(tái)受到了越來(lái)越多的關(guān)注。通過(guò)依靠Scala強(qiáng)有力的函數(shù)式編程、Actor通信模式、閉包、容器、泛型等手段，借助統(tǒng)一資源分配調(diào)度框架Mesos和YARN[2]，并融合了MapReduce和Dryad[3]等技術(shù)，使得Spark站在巨人的肩膀上，成為一個(gè)簡(jiǎn)潔、直觀、靈活、高效的分布式大數(shù)據(jù)處理框架。

本文對(duì)Spark中任務(wù)調(diào)度和資源分配機(jī)制進(jìn)行了研究，提出了一種基于計(jì)算節(jié)點(diǎn)同構(gòu)，即計(jì)算節(jié)點(diǎn)硬件環(huán)境一致下的資源分配優(yōu)化算法SMAO(Spark’s memory allocation optimizing)，能夠基于作業(yè)的數(shù)據(jù)量對(duì)分布式內(nèi)存資源進(jìn)行彈性分配，達(dá)到內(nèi)存資源合理分配的目的，并對(duì)此進(jìn)行了仿真實(shí)驗(yàn)驗(yàn)證。

二、Spark內(nèi)存計(jì)算框架

Spark在實(shí)時(shí)計(jì)算方面速度卓越的一個(gè)核心原因就是因?yàn)橛薪y(tǒng)一的RDD(Resilient Distributed Datasets，彈性分布式數(shù)據(jù)集)，Spark中使用了RDD抽象的分布式計(jì)算，即使用RDD對(duì)應(yīng)的transform/action等操作來(lái)執(zhí)行分布式計(jì)算；并且基于RDD之間的依賴關(guān)系組成lineage以及checkpoint等機(jī)制來(lái)保證整個(gè)分布式計(jì)算的容錯(cuò)性[4]。與Hadoop不同，Spark一開(kāi)始就瞄準(zhǔn)性能，將數(shù)據(jù)(包括中間結(jié)果)放在內(nèi)存中進(jìn)行計(jì)算，并且將用戶重復(fù)利用的數(shù)據(jù)也緩存到內(nèi)存中進(jìn)而提高計(jì)算效率，因此Spark尤其適合迭代型和交互性任務(wù)的計(jì)算，雖然Spark架構(gòu)下的計(jì)算需要大量的內(nèi)存，但其性能可隨著機(jī)器的數(shù)目呈現(xiàn)線性的增長(zhǎng)，其內(nèi)存計(jì)算速度比Hadoop的MapReduce要快100倍左右[5]。下面我們先看一下RDD，再對(duì)Spark運(yùn)行架構(gòu)進(jìn)行了解。

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

在Spark中一切運(yùn)算都是以RDD為基礎(chǔ)和核心的，而RDD的核心是RDD的分區(qū)存儲(chǔ)和RDD的有向無(wú)環(huán)圖的執(zhí)行[6]。在提交Job后，Spark依照對(duì)Job分解后的RDD之間的依賴關(guān)系形成Spark的調(diào)度順序，并結(jié)合任務(wù)的分發(fā)、跟蹤、執(zhí)行等過(guò)程，最終形成了整個(gè)的Spark應(yīng)用程序的執(zhí)行。

(二)Spark的應(yīng)用程序執(zhí)行機(jī)制

Spark應(yīng)用程序的執(zhí)行是基于SparkConf、SparkContext等環(huán)境信息的，在Spark 1.3.0中，可以人為設(shè)置參數(shù)spark.Executor.memory的大小來(lái)定義每一個(gè)Executor所需的內(nèi)存資源生成SparkConf配置變量，并通過(guò)val sc=new SparkContext(new SparkConf())來(lái)實(shí)例化該應(yīng)用程序的SparkContext上下文信息，Driver主控進(jìn)程會(huì)在SparkContext中構(gòu)建DAGScheduler等重要對(duì)象，負(fù)責(zé)對(duì)應(yīng)用進(jìn)行解析、stage切分并調(diào)度Task到Executor執(zhí)行。

Executor的運(yùn)行[8]是通過(guò)Spark的Akka消息機(jī)制向Master注冊(cè)程序后發(fā)送SparkConf配置信息給Worker進(jìn)而觸發(fā)啟動(dòng)的。在Master的消息響應(yīng)中會(huì)調(diào)用schedule方法和launchExecutor方法，在schedule方法中要啟動(dòng)Driver程序，然后調(diào)用Worker，而在launchExecutor內(nèi)部，Master會(huì)發(fā)送消息(消息即為L(zhǎng)aunchExecutor)給Worker節(jié)點(diǎn)讓它發(fā)起一個(gè)Executor，從Spark1.3.0的Executor消息處理源碼可以看出，ExecutorRunner負(fù)責(zé)分配Worker節(jié)點(diǎn)請(qǐng)求的CPU核數(shù)和Memory大小，實(shí)時(shí)地維護(hù)executor進(jìn)程，并負(fù)責(zé)executor的開(kāi)始和消亡。

三、采用SMAO對(duì)Executor內(nèi)存請(qǐng)求量進(jìn)行彈性分配

Executor執(zhí)行所占用內(nèi)存基本上是Executor內(nèi)部所有任務(wù)共享的，而每個(gè)Executor可以支持的任務(wù)的數(shù)量取決于Executor所管理的CPU Core資源和內(nèi)存資源的多少，目前的資源分配方式是是用戶了解每個(gè)任務(wù)的數(shù)據(jù)規(guī)模的大小，主觀地推算出每個(gè)Executor大致需要多少內(nèi)存和CPU。而在一般應(yīng)用中，Spark需要處理的數(shù)據(jù)量大小是變化的，比如本論文的基金項(xiàng)目中使用Spark來(lái)分析三峽庫(kù)區(qū)的水質(zhì)監(jiān)測(cè)數(shù)據(jù)，我們提出一種Spark計(jì)算節(jié)點(diǎn)同構(gòu)環(huán)境下的Executor內(nèi)存資源分配的優(yōu)化模型，能夠基于監(jiān)測(cè)數(shù)據(jù)量對(duì)Spark計(jì)算節(jié)點(diǎn)Executor所需的分布式內(nèi)存資源進(jìn)行彈性分配，達(dá)到內(nèi)存資源合理分配的目的，如下描述:

Executor所消耗的內(nèi)存，除了用于RDD數(shù)據(jù)集本身的開(kāi)銷，還包括算法所需各種臨時(shí)內(nèi)存空間的使用，即

SE≈SR+Sn

(1)

SE表示Executor所消耗的內(nèi)存，SR表示Executor中RDD數(shù)據(jù)集所占用的空間大小，Sn表示算法執(zhí)行所需要的各種空間，包括JVM的臨時(shí)空間消耗等各種計(jì)算消耗。默認(rèn)情況下，Spark采用整體物理內(nèi)存(spark.executor.memory)的60%來(lái)管理SR產(chǎn)生的RDD Cache數(shù)據(jù)，這表明在任務(wù)執(zhí)行期間，最多有40%的內(nèi)存可以給Sn用來(lái)保證任務(wù)運(yùn)行，用戶可以通過(guò)設(shè)置參數(shù)spark.storage.memoryFraction來(lái)改變這個(gè)比例大小。在使用Executor的過(guò)程中，如果數(shù)據(jù)量增大或分配的SR不夠則會(huì)報(bào)程序運(yùn)行空間不足，如果分配的Sn不夠，則會(huì)影響GC效率甚至報(bào)java.lang.OutOfMemoryError:Java heap space的錯(cuò)誤。用SF來(lái)表示頻繁發(fā)生Full GC時(shí)候的Sn的臨界值，在對(duì)Executor內(nèi)存資源分配時(shí)候中要保證以下條件：

k*SE≥SR,且(1-k)*SE≥SF

(2)

k表示參數(shù)spark.storage.memoryFraction，用戶可以自己設(shè)置這個(gè)參數(shù)來(lái)調(diào)整SR和Sn的比例。

用SF作為Sn的臨界值的時(shí)候，公式(1)變?yōu)?/p>

SE≈SR+SF

設(shè)X為原始數(shù)據(jù)的數(shù)據(jù)量，原始數(shù)據(jù)經(jīng)HDFS文件系統(tǒng)再轉(zhuǎn)換為Spark環(huán)境下的RDD數(shù)據(jù)集的轉(zhuǎn)換，RDD數(shù)據(jù)集所占空間即SR與原始數(shù)據(jù)的數(shù)據(jù)流大小呈現(xiàn)線性關(guān)系：

SR=p1X+p2

(3)

而算法執(zhí)行所需要的的開(kāi)銷Sn包含數(shù)據(jù)集本身的開(kāi)銷及各種算法的開(kāi)銷，算法不同，開(kāi)銷也不同，雖然這種開(kāi)銷難以統(tǒng)計(jì)。但是Sn與SR近似地呈現(xiàn)出Spark程序算法的空間復(fù)雜度的關(guān)系，也就意味著臨界值SF與原始數(shù)據(jù)X也近似地呈現(xiàn)Spark程序算法的空間復(fù)雜度的關(guān)系。對(duì)于SR，可以將RDD cache在內(nèi)存中，從Spark的log日志輸出或者UI輸出中可以看到每個(gè)Cache分區(qū)的大小，從而計(jì)算某數(shù)據(jù)量對(duì)應(yīng)的SR，對(duì)于SF，可以使用測(cè)試GC的方法來(lái)獲得。

本文提出一種Spark計(jì)算節(jié)點(diǎn)同構(gòu)情況下的Executor執(zhí)行時(shí)的內(nèi)存消耗優(yōu)化算法SMAO(Spark’s memory allocation optimizating)，計(jì)算步驟如下：

1.根據(jù)該應(yīng)用程序所運(yùn)行的作業(yè)量及RDD內(nèi)存消耗量的歷史記錄SR采用最小二乘法擬合得到公式(3)的參數(shù)p1和p2，進(jìn)而計(jì)算下一個(gè)作業(yè)的任務(wù)量X’對(duì)應(yīng)的SR’。

2.計(jì)算Spark所運(yùn)行的該應(yīng)用程序所消耗RDD的空間復(fù)雜度S(X)，根據(jù)該應(yīng)用程序所運(yùn)行的歷史作業(yè)的SF，進(jìn)而計(jì)算下一個(gè)作業(yè)X’，所產(chǎn)生的SF’。

3.由SR和SF代入公式1和公式2，計(jì)算SE’

4.在計(jì)算節(jié)點(diǎn)同構(gòu)情況下，對(duì)Spark提交下一個(gè)作業(yè)時(shí)候可以使用SE’的資源量作為每一個(gè)Executor的內(nèi)存分配請(qǐng)求，此時(shí)的SE’即Executor的內(nèi)存優(yōu)化分配的閾值。

此優(yōu)化算法相比沒(méi)有使用該算法優(yōu)化而言，使得Spark在處理數(shù)據(jù)時(shí)候隨著數(shù)據(jù)量的變化而彈性分配內(nèi)存資源，達(dá)到內(nèi)存資源合理分配的目的，下面將進(jìn)行仿真實(shí)驗(yàn)。

四、算法仿真與分析

(一)仿真實(shí)驗(yàn)環(huán)境及數(shù)據(jù)集

實(shí)驗(yàn)環(huán)境為由搭建的3個(gè)節(jié)點(diǎn)組成的虛擬機(jī)集群環(huán)境，使用ubuntu-12.04的64位版本的linux操作系統(tǒng)，產(chǎn)生一個(gè)主節(jié)點(diǎn)nameNode和兩個(gè)從節(jié)點(diǎn)dataNode1和dataNode2。虛擬機(jī)nameNode主節(jié)點(diǎn)的內(nèi)存為4GB，處理器數(shù)量為2，雙核，兩個(gè)dataNode內(nèi)存為2GB，處理器數(shù)量為2，單核。Java版本為JDK1.7.0_71，Hadoop的版本為2.4.0版本，Spark版本為1.3.0，實(shí)驗(yàn)的程序是根據(jù)Spark自帶的PCA程序而改寫的帶有標(biāo)準(zhǔn)化處理的PCA程序(SNPCA,Spark’s Normalized Principal Component Analysis)。

前面講過(guò)，在Executor中所消耗的內(nèi)存，除了用于RDD數(shù)據(jù)集本身的開(kāi)銷外，還包括算法所需各種臨時(shí)內(nèi)存空間Sn的使用，SF表示頻繁發(fā)生Full GC時(shí)候的Sn的臨界值，SF受系統(tǒng)環(huán)境影響，我們可以假設(shè)實(shí)驗(yàn)程序SNPCA引起的SF開(kāi)銷符合以下空間復(fù)雜度：

S(x)=q1x3+q2x2+q3x+q4

(4)

使用最小二乘法對(duì)SF的歷史數(shù)據(jù)來(lái)進(jìn)行多項(xiàng)式擬合，進(jìn)而計(jì)算后三個(gè)測(cè)試集數(shù)據(jù)[131,143,155]MB實(shí)際的SF量為[361,379,482]MB，而依照SMAO算法計(jì)算得[414.9,498.5,599.1]MB，對(duì)比如下：

實(shí)際的SF量與依據(jù)SMAO算法計(jì)算的SF量對(duì)比圖

使用最小二乘法對(duì)SF的歷史數(shù)據(jù)來(lái)進(jìn)行多項(xiàng)式擬合的結(jié)果，即公式4的系數(shù)為：[0.0002,-0.0347,4.0694,-17.0236]，也能看出來(lái)實(shí)驗(yàn)程序SNPCA算法的空間復(fù)雜度基本上還是一個(gè)線性函數(shù)。

計(jì)算出SR量和SF量后，可以依照公式1、公式2計(jì)算計(jì)算節(jié)點(diǎn)同構(gòu)環(huán)境下的SE’，但是對(duì)于實(shí)驗(yàn)的SNPCA程序，Executor中SF開(kāi)銷與原始數(shù)據(jù)量呈線性關(guān)系，并且SR與數(shù)據(jù)量X(公式1)也是線性關(guān)系，所以此時(shí)Executor的分配量與數(shù)據(jù)量也呈線性關(guān)系，所以仍然可以使用公式4對(duì)數(shù)據(jù)量X’和Executor的開(kāi)銷SE’進(jìn)行分析，為充分起見(jiàn)，可以對(duì)Executor的內(nèi)存分配“盈余”一定比例，比如“盈余”0.1時(shí)，取Executor內(nèi)存優(yōu)化分配量為1.1*SE’，來(lái)充分提供除

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

理論上給Executor的內(nèi)存分配當(dāng)然是多多益善，實(shí)際受機(jī)器配置、以及運(yùn)行環(huán)境、資源共享、JVM GC效率等因素的影響難以具體衡量，但是還是可以使用SR和SF的歷史數(shù)據(jù)來(lái)進(jìn)行內(nèi)存合理分配。

從以上可以看出，依據(jù)SMAO算法計(jì)算的SR與實(shí)際大小相比幾乎一樣，Spark在作業(yè)處理時(shí)候的內(nèi)存消耗量除了RDD數(shù)據(jù)集本身的SR開(kāi)銷之外，還包括所需各種臨時(shí)內(nèi)存空間的使用帶來(lái)的其他消耗量SF。

需要注意的是，SMAO算法目前應(yīng)用于計(jì)算節(jié)點(diǎn)同構(gòu)的情況下，SRAO算法的使用需要預(yù)先知道Spark應(yīng)用程序產(chǎn)生RDD數(shù)據(jù)集的空間復(fù)雜度，因此對(duì)于不同的作業(yè)程序，需要人為地事先計(jì)算空間復(fù)雜度，另外，如果Executor執(zhí)行的時(shí)候頻繁發(fā)生Full GC，可以考慮減小這個(gè)spark.storage.memoryFraction比值來(lái)減少Full GC發(fā)生的次數(shù)，來(lái)改善程序運(yùn)行的整體性能。

五、結(jié)束語(yǔ)

本文首先對(duì)Spark的架構(gòu)和應(yīng)用執(zhí)行機(jī)制進(jìn)行了介紹，并引出了Spark 1.3.0在Executor的內(nèi)存分配方面還不太令人滿意，目前采用的是人為的任務(wù)分配機(jī)制，針對(duì)此問(wèn)題，提出了基于Spark的同構(gòu)計(jì)算節(jié)點(diǎn)環(huán)境下Executor的內(nèi)存分配優(yōu)化算法，并展開(kāi)實(shí)驗(yàn)驗(yàn)證該優(yōu)化方案，實(shí)驗(yàn)結(jié)果顯示優(yōu)化方案的有效性。

[1]http://spark.apache.org/.Accessed Oct,2015

[2]Hindman B,Konwinski A,Zaharia M,et al.Mesos:A Platform for Fine-Grained Resource Sharing in the Data Center[C].NSDI.2011,11:22-22.

[3]Isard M,Budiu M,Yu Y,et al.Dryad:distributed data-parallel programs from sequential building blocks[C].ACM SIGOPS Operating Systems Review.ACM,2007,41(3):59-72

[4]Zaharia M,Chowdhury M,Das T,et al.Resilient distributed datasets:A fault-tolerant abstraction for in-memory cluster computing[C]//Proceedings of the 9th USENIX conference on Networked Systems Design and Implementation.USENIX Association,2012:2-2.

[5]Zaharia M,Chowdhury M,Franklin M J,et al.Spark:cluster computing with working sets[C]//Proceedings of the 2nd USENIX conference on Hot topics in cloud computing.2010:10-10.

[6]Zaharia M,Chowdhury M,Das T,et al.Resilient distributed datasets:A fault-tolerant abstraction for in-memory cluster computing[C].Proceedings of the 9th USENIX conference on Networked Systems Design and Implementation.USENIX Association,2012:2-2.

[7]高彥杰.Spark大數(shù)據(jù)處理技術(shù)、應(yīng)用與性能優(yōu)化[M].北京：機(jī)械工業(yè)出版社，2014

[8]王家林.大數(shù)據(jù)Spark企業(yè)級(jí)實(shí)戰(zhàn)[M].北京：電子工業(yè)出版社，2015