卞 琛，于 炯，修位蓉，英昌甜，錢(qián)育蓉

(新疆大學(xué) 信息科學(xué)與工程學(xué)院，烏魯木齊 830046) (*通信作者電子郵箱bianchen0720@126.com)

基于迭代填充的內(nèi)存計(jì)算框架分區(qū)映射算法

卞 琛*，于 炯，修位蓉，英昌甜，錢(qián)育蓉

(新疆大學(xué) 信息科學(xué)與工程學(xué)院，烏魯木齊 830046) (*通信作者電子郵箱bianchen0720@126.com)

針對(duì)內(nèi)存計(jì)算框架Spark在作業(yè)Shuffle階段一次分區(qū)產(chǎn)生的數(shù)據(jù)傾斜問(wèn)題，提出一種內(nèi)存計(jì)算框架的迭代填充分區(qū)映射算法(IFPM)。首先,分析Spark作業(yè)的執(zhí)行機(jī)制，建立作業(yè)效率模型和分區(qū)映射模型，給出作業(yè)執(zhí)行時(shí)間和分配傾斜度的定義，證明這些定義與作業(yè)執(zhí)行效率的因果邏輯關(guān)系;然后,根據(jù)模型和定義求解，設(shè)計(jì)擴(kuò)展式數(shù)據(jù)分區(qū)算法(EPA)和迭代式分區(qū)映射算法(IMA)，在Map端建立一對(duì)多分區(qū)函數(shù)，并通過(guò)分區(qū)函數(shù)將部分?jǐn)?shù)據(jù)填入擴(kuò)展區(qū)內(nèi)，在數(shù)據(jù)分布局部感知后再執(zhí)行擴(kuò)展區(qū)迭代式的多輪數(shù)據(jù)分配，根據(jù)Reduce端已分配數(shù)據(jù)量建立適應(yīng)性的擴(kuò)展區(qū)映射規(guī)則，對(duì)原生區(qū)的數(shù)據(jù)傾斜進(jìn)行逐步修正，以此保障數(shù)據(jù)分配的均衡性。實(shí)驗(yàn)結(jié)果表明，在不同源數(shù)據(jù)分布條件下，算法均提高了作業(yè)Shuffle過(guò)程分區(qū)映射合理性，縮減了寬依賴(lài)Stage的同步時(shí)間，提高了作業(yè)執(zhí)行效率。

內(nèi)存計(jì)算；數(shù)據(jù)均衡；擴(kuò)展式分區(qū)；迭代式映射

0 引言

近年來(lái)，利用內(nèi)存的低延遲特性改進(jìn)并行計(jì)算框架性能成為新的研究方向。內(nèi)存計(jì)算框架避免了頻繁訪(fǎng)問(wèn)磁盤(pán)的I/O性能瓶頸，解放了大內(nèi)存+多核處理器硬件架構(gòu)的潛在高性能，成為學(xué)術(shù)界一致認(rèn)可的高性能并行計(jì)算系統(tǒng)[1-2]。雖然內(nèi)存計(jì)算框架的性能表現(xiàn)優(yōu)異，但與大數(shù)據(jù)時(shí)代的即時(shí)應(yīng)用需求相比，還存在不小的差距，因此，從計(jì)算模型的角度研究?jī)?nèi)存計(jì)算框架的性能優(yōu)化方法具有一定的現(xiàn)實(shí)意義。本文選取開(kāi)源內(nèi)存計(jì)算框架Spark[3-4]為研究對(duì)象，Spark以HDFS(Hadoop Distributed File System)為底層文件系統(tǒng)，采用彈性分布式數(shù)據(jù)集(Resilient Distributed Datasets, RDD)[5]作為數(shù)據(jù)結(jié)構(gòu)，通過(guò)數(shù)據(jù)集血統(tǒng)(lineage)[5-6]和檢查點(diǎn)機(jī)制(checkpoint)[7-8]實(shí)現(xiàn)系統(tǒng)容錯(cuò)，編程模式則借鑒了函數(shù)式編程語(yǔ)言的設(shè)計(jì)思想，簡(jiǎn)化了多階段作業(yè)的流程跟蹤、任務(wù)重新執(zhí)行和周期性檢查點(diǎn)機(jī)制的實(shí)現(xiàn)。

在Spark作業(yè)的寬依賴(lài)Stage執(zhí)行過(guò)程中，Mapper將數(shù)據(jù)按key劃分并填入不同的Bucket，Bucket與Reducer為一一對(duì)應(yīng)關(guān)系。由于原始數(shù)據(jù)分布的傾斜性，這樣的單一輪次分區(qū)映射過(guò)程使各Reducer計(jì)算數(shù)據(jù)量有較大差異，任務(wù)執(zhí)行時(shí)間長(zhǎng)短不一，從而增加了寬依賴(lài)Stage的計(jì)算延時(shí)，降低了作業(yè)執(zhí)行效率。雖然系統(tǒng)支持用戶(hù)設(shè)定自定義分區(qū)函數(shù)，但由于真實(shí)的數(shù)據(jù)分布難以預(yù)知，無(wú)法確保自定義分區(qū)函數(shù)的合理性和準(zhǔn)確性，因此數(shù)據(jù)分配的傾斜問(wèn)題不可規(guī)避。為解決這一問(wèn)題，本文主要做了以下工作：

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

2)通過(guò)分析寬依賴(lài)RDD的計(jì)算過(guò)程，建立了分區(qū)映射模型，給出了源數(shù)據(jù)分布、分區(qū)映射、分配傾斜度的定義，并證明這些定義與作業(yè)執(zhí)行效率的因果關(guān)系。

3)通過(guò)模型的相關(guān)定義求解，設(shè)計(jì)了擴(kuò)展式數(shù)據(jù)分區(qū)算法和迭代式分區(qū)映射算法，并對(duì)算法執(zhí)行的細(xì)節(jié)問(wèn)題進(jìn)行詳細(xì)的分析和說(shuō)明。

1 相關(guān)工作

在提出MapReduce的文獻(xiàn)[9]中，Dean等采用Hash函數(shù)對(duì)數(shù)據(jù)進(jìn)行一次簡(jiǎn)單的劃分，由于這種方法實(shí)現(xiàn)簡(jiǎn)單且通用性高，成為開(kāi)源的Hadoop系統(tǒng)默認(rèn)的分區(qū)方案。Spark作為類(lèi)MapReduce系統(tǒng)，在實(shí)現(xiàn)中也自然承接了MapReduce的分區(qū)方法，但實(shí)際應(yīng)用表明，在不了解數(shù)據(jù)分布的情況下，一次Hash劃分的方法很難實(shí)現(xiàn)數(shù)據(jù)的合理分配。

一些研究成果致力于通過(guò)優(yōu)化原生的分區(qū)映射策略解決數(shù)據(jù)分配的均衡性問(wèn)題，文獻(xiàn)[10]研究Map和Reduce兩個(gè)階段的任務(wù)執(zhí)行過(guò)程，通過(guò)分析數(shù)據(jù)不均衡分配的原因，歸納出數(shù)據(jù)傾斜的5個(gè)類(lèi)別。文獻(xiàn)[11]提出SkewReduce策略，該策略建立用戶(hù)定義的代價(jià)模型，在作業(yè)執(zhí)行過(guò)程逐步收集元數(shù)據(jù)，鄰近代價(jià)閾值時(shí)啟動(dòng)分區(qū)映射過(guò)程，以實(shí)現(xiàn)計(jì)算數(shù)據(jù)量的均勻分配。文獻(xiàn)[12]提出MapReduce的增量式分區(qū)策略，將原始數(shù)據(jù)劃分為細(xì)粒度的微分區(qū)，通過(guò)數(shù)據(jù)分布的逐步感知和已分配數(shù)據(jù)量的統(tǒng)計(jì)，采用Max-Min算法進(jìn)行數(shù)據(jù)增量分配，達(dá)到數(shù)據(jù)分配逐漸均衡的目標(biāo)。文獻(xiàn)[13]提出SkewTune，與上述的研究成果不同，SkewTune建立Reducer的任務(wù)剩余代價(jià)評(píng)估模型，通過(guò)對(duì)Reducer執(zhí)行進(jìn)度進(jìn)行統(tǒng)計(jì)，決定是否將數(shù)據(jù)向其他Reducer遷移。由于數(shù)據(jù)的二次遷移將延遲Reducer計(jì)算任務(wù)，因此相比設(shè)計(jì)分區(qū)策略保證數(shù)據(jù)均衡分配的方法，SkewTune具有較大的額外開(kāi)銷(xiāo)。文獻(xiàn)[14]為實(shí)現(xiàn)分區(qū)數(shù)據(jù)的實(shí)時(shí)統(tǒng)計(jì)，在系統(tǒng)中增加額外的數(shù)據(jù)構(gòu)Sketch-based，通過(guò)設(shè)計(jì)的分包算法進(jìn)行Reducer計(jì)算數(shù)據(jù)量的動(dòng)態(tài)調(diào)配，達(dá)到數(shù)據(jù)均衡分配的目標(biāo)。

另外一些研究成果期望通過(guò)數(shù)據(jù)分布的逐步感知建立合理的分區(qū)映射方案。文獻(xiàn)[15]通過(guò)在Mapper增加采樣進(jìn)程感知原始數(shù)據(jù)分布，已生成的分區(qū)容量達(dá)到閾值后進(jìn)行重組或拆分，保障分配數(shù)據(jù)的均衡性。文獻(xiàn)[16-17]提出精細(xì)分區(qū)和動(dòng)態(tài)拆分兩種算法，精細(xì)分區(qū)算法采樣獲得近似數(shù)據(jù)分布，動(dòng)態(tài)拆分函數(shù)在Map任務(wù)完成一定比例后觸發(fā)，進(jìn)行分區(qū)容量的二次調(diào)整，達(dá)到數(shù)據(jù)合理分配的目標(biāo)。文獻(xiàn)[18-19]提出基于〈block,entity〉數(shù)據(jù)塊的分區(qū)方法，通過(guò)評(píng)估函數(shù)對(duì)超出閾值的數(shù)據(jù)塊進(jìn)行調(diào)整，但沒(méi)有精確定義分區(qū)調(diào)整的時(shí)機(jī)問(wèn)題。文獻(xiàn)[20]提出提前采樣的策略，在Map任務(wù)執(zhí)行前先對(duì)輸入數(shù)據(jù)進(jìn)行25%的隨機(jī)采樣，通過(guò)采樣結(jié)果獲得數(shù)據(jù)分布并制定分區(qū)函數(shù)。文獻(xiàn)[21]提出LEEN策略，通過(guò)對(duì)輸入數(shù)據(jù)的預(yù)掃描獲取數(shù)據(jù)分布，在Map任務(wù)執(zhí)行過(guò)程中逐步統(tǒng)計(jì)key的頻率，然后綜合數(shù)據(jù)分布和key頻率設(shè)定合理的分區(qū)函數(shù)。

本文與上述研究成果的不同之處在于從寬依賴(lài)Stage數(shù)據(jù)分配的基本原理入手，以提高作業(yè)的整體執(zhí)行效率為目的，設(shè)計(jì)了迭代填充分區(qū)映射算法，解決同構(gòu)集群環(huán)境下數(shù)據(jù)分配的均衡性問(wèn)題。通過(guò)分析作業(yè)的執(zhí)行過(guò)程，建立了作業(yè)效率模型，提出了RDD計(jì)算代價(jià)和作業(yè)執(zhí)行時(shí)間的定義。建立分區(qū)映射模型，提出了源數(shù)據(jù)分布、分配傾斜度的定義，并證明了兩個(gè)定義與作業(yè)執(zhí)行時(shí)間的因果關(guān)系。根據(jù)模型和定義求解，設(shè)計(jì)了擴(kuò)展式數(shù)據(jù)分區(qū)算法和迭代式分區(qū)映射算法。通過(guò)擴(kuò)展式分區(qū)預(yù)留部分原始數(shù)據(jù)，并設(shè)計(jì)擴(kuò)展區(qū)的延遲映射機(jī)制，為迭代式分區(qū)映射奠定基礎(chǔ)。通過(guò)擴(kuò)展區(qū)迭代式的多輪數(shù)據(jù)分配，對(duì)原生區(qū)的數(shù)據(jù)傾斜進(jìn)行逐步修正，減少各Reducer分配數(shù)據(jù)量差異，從而從整體上提高寬依賴(lài)Stage的計(jì)算速度，提高作業(yè)執(zhí)行效率。相比已有的研究工作，迭代填充分區(qū)映射算法更適宜于內(nèi)存計(jì)算框架的性能優(yōu)化，并具有較高的普適性和易用性。

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

本章首先分析Spark作業(yè)的執(zhí)行機(jī)制，建立作業(yè)效率模型和分區(qū)映射模型，然后提出迭代填充分區(qū)映射的優(yōu)化目標(biāo)，為第3章的算法設(shè)計(jì)提供理論基礎(chǔ)。

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

Spark將操作分為T(mén)ransformation和Action兩類(lèi)，調(diào)度策略采用延時(shí)調(diào)度機(jī)制，即當(dāng)Action操作執(zhí)行時(shí)，作業(yè)才會(huì)分發(fā)到集群執(zhí)行?；谘訒r(shí)調(diào)度的原理，Spark會(huì)首先根據(jù)RDD的血統(tǒng)生成作業(yè)的有向無(wú)環(huán)圖(Directed Acyclic Graph，DAG)，如圖1所示。其中虛線(xiàn)框代表Stage，圓角矩形代表RDD，填充方框表示RDD分區(qū)。Stage的劃分以寬依賴(lài)為邊界，各Stage順序執(zhí)行，直至計(jì)算出最終結(jié)果。集群任務(wù)分配則以數(shù)據(jù)本地性作為依據(jù)，即任務(wù)總是調(diào)度給具有最佳數(shù)據(jù)本地性的工作節(jié)點(diǎn)，以減少網(wǎng)絡(luò)通信延時(shí)，提高作業(yè)執(zhí)行效率。

圖1 Spark作業(yè)的有向無(wú)環(huán)圖

2.2 作業(yè)效率模型

根據(jù)2.1節(jié)的描述，Spark作業(yè)在執(zhí)行時(shí)劃分為多個(gè)Stage同步執(zhí)行，每個(gè)Stage由一個(gè)或多個(gè)RDD構(gòu)成，每個(gè)RDD由多個(gè)分區(qū)并行計(jì)算生成，因此，記一個(gè)作業(yè)的Stage集合為stages={stg1,stg2,…,stgi}，每個(gè)Stage包含的RDD表示為集合stgi={RDDi1,RDDi2,…,RDDij}， 其中RDDij表示第i個(gè)Stage中第j個(gè)RDD，對(duì)于每個(gè)RDD，其分區(qū)集合記為RDDij={Pij1,Pij2,…,Pijk}，這里Pijk表示RDDij中的第k個(gè)分區(qū)。

定義1 RDD計(jì)算代價(jià)。Spark任務(wù)中，分區(qū)是最基本的計(jì)算單位，分區(qū)計(jì)算首先要讀取輸入，再根據(jù)閉包運(yùn)算符和操作符進(jìn)行運(yùn)算。設(shè)Parentsijk為分區(qū)Pijk的父分區(qū)集合，用于表示分區(qū)計(jì)算的輸入數(shù)據(jù)，那么分區(qū)Pijk的計(jì)算代價(jià)為數(shù)據(jù)讀取代價(jià)與數(shù)據(jù)處理代價(jià)之和，本文以分區(qū)計(jì)算時(shí)間作為衡量計(jì)算代價(jià)的唯一指標(biāo)，即：

TPijk=read(Parentsijk)+proc(Parentsijk)

(1)

每個(gè)RDD的分區(qū)分配到不同的工作節(jié)點(diǎn)并行計(jì)算生成，因此RDD計(jì)算代價(jià)為所有分區(qū)計(jì)算代價(jià)的最大值，即：

TRDDij=max(TPij1,TPij2,…,TPijk)

(2)

定義2 作業(yè)執(zhí)行時(shí)間。如圖1所示，Spark將Stage分為窄依賴(lài)和寬依賴(lài)兩類(lèi)。對(duì)于窄依賴(lài)Stage,每個(gè)Stage包括多條流水線(xiàn)(每條流水線(xiàn)包括多個(gè)RDD的不同分區(qū))。設(shè)窄依賴(lài)stagei共有h個(gè)RDD，所有RDD劃分為x條流水線(xiàn)，單條流水線(xiàn)的分區(qū)集合為pipeix={Pi1x,Pi2x,…,Pijx}，那么單條流水線(xiàn)的執(zhí)行時(shí)間可表示為：

(3)

對(duì)于stagei，記其流水線(xiàn)集合為Pipesi={pipei1,pipei2,…,pipeix}，那么stagei的執(zhí)行時(shí)間應(yīng)為各流水線(xiàn)執(zhí)行時(shí)間最大值，即：

Tstagei=max(Tpipei1,Tpipei2,…,Tpipeix)

(4)

設(shè)stagei+1為寬依賴(lài)，則其中僅包含一個(gè)RDD的計(jì)算任務(wù)，記為RDD(i+1)j，那么stagei的執(zhí)行時(shí)間與RDD(i+1)j的計(jì)算代價(jià)相同，即：

Tstagei+1=TRDD(i+1)j

(5)

若Spark作業(yè)共有n個(gè)Stage(其中包括若干個(gè)窄依賴(lài)和寬依賴(lài)Stage)，則各Stage順序執(zhí)行，因此作業(yè)執(zhí)行總時(shí)長(zhǎng)為：

(6)

2.3 分區(qū)映射模型

作業(yè)的寬依賴(lài)Stage分Map和Reduce兩個(gè)階段執(zhí)行，其中Map階段將前一Stage的生成結(jié)果轉(zhuǎn)化為〈key,value〉元組，放入不同的Bucket中，每個(gè)Bucket對(duì)應(yīng)一個(gè)Reduce任務(wù)，所有Map任務(wù)執(zhí)行結(jié)束后，由Reducer到各個(gè)工作節(jié)點(diǎn)拉取對(duì)應(yīng)Bucket的數(shù)據(jù)，完成后續(xù)計(jì)算。由于工作節(jié)點(diǎn)內(nèi)存空間有限，為防止頻繁內(nèi)存回收，Spark將Bucket數(shù)據(jù)寫(xiě)入磁盤(pán)，以保證Reducer輸入數(shù)據(jù)的可用性。

定義3 源數(shù)據(jù)分布。用于描述輸入數(shù)據(jù)在Mapper端的分布情況。記源數(shù)據(jù)的key集合為keys={key1,key2,…,keyl}，即源數(shù)據(jù)有l(wèi)個(gè)不同的key，記作業(yè)的Mapper集合為mps={1, 2,…,m}，那么對(duì)于編號(hào)為m的任意Mapper，其數(shù)據(jù)分布可表示為：

Am=(Am1,Am2,…,Aml)T

(7)

其中Aml表示第l個(gè)key在第m個(gè)Mapper上的數(shù)據(jù)量。將所有Mapper的數(shù)據(jù)分布向量進(jìn)行歸并，那么源數(shù)據(jù)的整體分布可表示為m×l矩陣：

(8)

矩陣中同行元素表示相同key在不同Mapper上的數(shù)據(jù)分布，映射過(guò)程同行元素也由相同的Reducer完成計(jì)算，因此將數(shù)據(jù)按key進(jìn)行數(shù)據(jù)量統(tǒng)計(jì)，任意key的數(shù)據(jù)總量可表示為：

(9)

那么將源數(shù)據(jù)按key進(jìn)行劃分，可表示為如下集合：

S={c1,c2,…,cl};l∈keys

(10)

定義4 分區(qū)映射。用于描述Mapper數(shù)據(jù)分布中key與Reducer之間的映射關(guān)系，分區(qū)映射也表示與Reducer對(duì)應(yīng)Bucket的填充規(guī)則。Spark系統(tǒng)延用MapReduce的一次分區(qū)機(jī)制，默認(rèn)對(duì)key進(jìn)行哈希值轉(zhuǎn)換，再與Reducer的數(shù)量取模，以此決定數(shù)據(jù)所對(duì)應(yīng)的Bucket，因此原生的分區(qū)函數(shù)可表示為：

f(Bucket)=hash(key)mod(n)

(11)

通過(guò)上述的分區(qū)函數(shù)可以看出，分區(qū)函數(shù)保證相同key值數(shù)據(jù)存放在同一個(gè)Bucket。由于所有Mapper采用同一分區(qū)函數(shù)劃分?jǐn)?shù)據(jù)，因此源數(shù)據(jù)中所有相同key數(shù)據(jù)都映射到同一Reducer。

記作業(yè)的Reducer集合為rds={rd1,rd2,…,rdn}，那么任意Reducer的分區(qū)映射關(guān)系可表示為：

inputrdi|→ {ci,cn+i,…,cj×n+i};j∈[0,l/n]

(12)

定義5 分配傾斜度。用于描述Reducer分配數(shù)據(jù)與均值的差異程度。由定義3可知，Reducer集合要處理的數(shù)據(jù)總量為S，那么在同構(gòu)集群環(huán)境下，各Reducer分配數(shù)據(jù)量的均值應(yīng)表示為：

(13)

根據(jù)定義4的描述，Spark依舊延用MapReduce的一次分區(qū)技術(shù)，將key的哈希值與Reducer數(shù)量取模，以判定該key數(shù)據(jù)與Reducer的對(duì)應(yīng)關(guān)系，但由于key的哈希值與其在數(shù)據(jù)分布的出現(xiàn)頻率無(wú)關(guān)，即與相同key的元組數(shù)無(wú)關(guān)，因此在多數(shù)情況下，各Reducer的數(shù)據(jù)分配量與均值不匹配，根據(jù)式(11)、(12)，將任意Reducer的分配傾斜度定義為：

Qi=inputrdi/E

(14)

定理1 在同構(gòu)集群環(huán)境下，對(duì)于所有執(zhí)行寬依賴(lài)Stage的Reducer，其分配傾斜度越小，作業(yè)的執(zhí)行效率越高。

證明 設(shè)作業(yè)當(dāng)前執(zhí)行寬依賴(lài)Stagei，基于定義3，寬依賴(lài)Stage僅包含一個(gè)RDD的計(jì)算工作，因此Stagei的執(zhí)行時(shí)間等于RDDij的計(jì)算代價(jià)。記任意的Reducer計(jì)算時(shí)間為T(mén)finishn，由于在任務(wù)分配中，每個(gè)Reducer負(fù)責(zé)計(jì)算RDDij的一個(gè)分區(qū)，因此Stagei的執(zhí)行時(shí)間也可表示為：

Tstagei=TRDDij=max(Tfinish1,Tfinish2,…Tfinishn)

(15)

同構(gòu)集群環(huán)境下，各Reducer的計(jì)算能力基本一致，因此輸入數(shù)據(jù)量成為決定計(jì)算時(shí)長(zhǎng)的唯一因素。根據(jù)定義5的描述，分配傾斜度表示Reducer數(shù)據(jù)分配與均值的差異，均值代表完全均勻的數(shù)據(jù)分配，因此對(duì)于所有執(zhí)行寬依賴(lài)Stage的Reducer，其分配傾斜度越小，作業(yè)的執(zhí)行效率越高。

3 迭代填充分區(qū)映射算法

本章基于模型相關(guān)的定義和證明，提出擴(kuò)展式數(shù)據(jù)分區(qū)算法和迭代式分區(qū)映射算法，并對(duì)算法的執(zhí)行細(xì)節(jié)進(jìn)行分析和說(shuō)明。

3.1 算法的總體描述

根據(jù)2.3節(jié)定義4，傳統(tǒng)Spark的分區(qū)方法延用MapReduce的一次劃分方法，數(shù)據(jù)分配與key的個(gè)數(shù)有關(guān)而與分配數(shù)據(jù)量無(wú)關(guān)，導(dǎo)致數(shù)據(jù)發(fā)生傾斜影響作業(yè)執(zhí)行效率，因此，迭代填充分區(qū)映射算法的目標(biāo)是提高數(shù)據(jù)分配策略與數(shù)據(jù)量的關(guān)聯(lián)度，以此增加分配策略的合理性，但由于在所有Map任務(wù)完成之前難以預(yù)知真實(shí)的數(shù)據(jù)分布，因此考慮改進(jìn)既有的一次分區(qū)策略，通過(guò)多輪的分區(qū)映射過(guò)程達(dá)到數(shù)據(jù)適應(yīng)性分配的目標(biāo)。

迭代填充分區(qū)映射算法的主要思想是：1)將Mapper與Reducer之間的數(shù)據(jù)緩沖區(qū)劃分為原生區(qū)和擴(kuò)展區(qū)兩部分，每個(gè)區(qū)域包含的Bucket數(shù)量與Reducer的個(gè)數(shù)相同。2)在原生分區(qū)策略的基礎(chǔ)上加以改進(jìn)，保證大部分?jǐn)?shù)據(jù)寫(xiě)入原生區(qū)，而小部分key的數(shù)據(jù)寫(xiě)入擴(kuò)展區(qū)，并能夠?qū)?yīng)到擴(kuò)展區(qū)中不同的Bucket編號(hào)。3)原生區(qū)中的Bucket與Reducer之間為固定對(duì)應(yīng)關(guān)系，當(dāng)某個(gè)Mapper計(jì)算完畢后，所有Reducer即可開(kāi)始進(jìn)行原生區(qū)的數(shù)據(jù)拉取。4)初始狀態(tài)下，擴(kuò)展區(qū)中的Bucket與Reducer無(wú)對(duì)應(yīng)關(guān)系，達(dá)到特定時(shí)機(jī)則啟動(dòng)后續(xù)輪次分配，將擴(kuò)展區(qū)的數(shù)據(jù)逐步映射到Reducer。

3.2 擴(kuò)展式數(shù)據(jù)分區(qū)算法

擴(kuò)展式數(shù)據(jù)分區(qū)算法的主要步驟如下：

1)確定擴(kuò)展參數(shù)x，原生區(qū)和擴(kuò)展區(qū)生成Bucket，原生區(qū)的Bucket數(shù)量與Reducer的個(gè)數(shù)n相同，擴(kuò)展區(qū)的Bucket數(shù)量為n×x。

2)Mapper計(jì)算hash(key)mod(n+x)獲得寫(xiě)入數(shù)據(jù)的Bucket編號(hào)，若編號(hào)小于n，則寫(xiě)入數(shù)據(jù)，本次過(guò)程結(jié)束；若編號(hào)大于等于n，則表示數(shù)據(jù)應(yīng)放入擴(kuò)展區(qū)，繼續(xù)執(zhí)行步驟3)。

3)對(duì)于hash(key)mod(n+x)≥n的情況，繼續(xù)計(jì)算(hash(key)/(n+x))mod(n×x)，確定該數(shù)據(jù)在擴(kuò)展區(qū)中的Bucket編號(hào)并寫(xiě)入數(shù)據(jù)，本次過(guò)程結(jié)束。

擴(kuò)展式數(shù)據(jù)分區(qū)算法的偽代碼如下：

算法1 擴(kuò)展式數(shù)據(jù)分區(qū)算法。

輸入：原生區(qū)native;擴(kuò)展區(qū)extension;Reducer個(gè)數(shù)n;源數(shù)據(jù)鍵值key;擴(kuò)展參數(shù)x。

初始化：bukNo←-1;

1)

native.creatBucket(n);

2)

extension.creatBucket(n*x);

3)

bukNo=hash(key) mod (n+x);

4)

if(bukNo

5)

write(key,native[bukNo]);

6)

else

7)

bukNo= (hash(key) /(n+x)) mod (n*x);

8)

write(key,extension[bukNo]);

9)

end if

由算法描述可以看出，擴(kuò)展參數(shù)決定了原生區(qū)與擴(kuò)展區(qū)的劃分比例，而擴(kuò)展區(qū)則為后續(xù)的分區(qū)映射算法服務(wù)，通過(guò)多輪分配漸進(jìn)填充，提高數(shù)據(jù)分配的合理性。

3.3 迭代式分區(qū)映射算法

根據(jù)3.1節(jié)的描述，原生區(qū)的Bucket數(shù)量與Reducer個(gè)數(shù)相同，兩者之間為一一對(duì)應(yīng)關(guān)系，由于原生區(qū)的生成方式與MapReduce的一次分區(qū)策略相同，難以保證數(shù)據(jù)的均勻分配，因而擴(kuò)展區(qū)的后續(xù)輪次分配的合理性成為算法目標(biāo)實(shí)現(xiàn)的關(guān)鍵問(wèn)題。為達(dá)到精準(zhǔn)分配，本文方法在原生Spark系統(tǒng)中增加了1個(gè)計(jì)數(shù)器counter和1個(gè)數(shù)據(jù)構(gòu)RelationSchema，counter用于統(tǒng)計(jì)擴(kuò)展區(qū)內(nèi)各Bucket的數(shù)據(jù)量，RelationSchema用于表示Bucket與Reducer的映射關(guān)系。原生區(qū)映射過(guò)程與傳統(tǒng)Spark相同，不再贅述，下面重點(diǎn)討論擴(kuò)展區(qū)的映射過(guò)程，其主要步驟如下：

1)將擴(kuò)展區(qū)中的Bucket倒序排列，并選取前n個(gè)Bucket生成待分配列表。

2)對(duì)所有Reducer的RelationSchema進(jìn)行映射數(shù)據(jù)量統(tǒng)計(jì)，挑選出映射數(shù)據(jù)量最小的Reducer。

3)將分配列表容量最大的Bucket與映射數(shù)據(jù)量最小的Reducer建立一一對(duì)應(yīng)的映射關(guān)系，更新RelationSchema。

4)重復(fù)步驟2)，直到n個(gè)Bucket都映射完畢。

5)啟動(dòng)數(shù)據(jù)拉取進(jìn)程，等待下一輪映射過(guò)程。

算法2 迭代式分區(qū)映射算法。

輸入：擴(kuò)展區(qū)extension;Reducer集合rds;

初始化：candis←newList〈Bucket〉;

//待分配列表

1)

extension.orderDesc();

2)

candis=extension.getTop(n);

3)

fori=0 ton-1 do

4)

rds.RelationSchema.statistics();

5)

minload=min(rds);

//負(fù)載最小Reducer

6)

minload.mapping(candis[i]);

//建立映射

7)

minload.RelationSchema.update();

8)

end for

9)

start pull;

//啟動(dòng)數(shù)據(jù)拉取進(jìn)程

10)

waitfor nextround;

//等待下一輪分配

接下來(lái)討論分區(qū)映射算法執(zhí)行的時(shí)機(jī)問(wèn)題，原生區(qū)的映射過(guò)程依舊采用傳統(tǒng)Spark的處理方式，即當(dāng)?shù)?個(gè)Mapper計(jì)算完成后，所有Reducer即可從該Mapper拉取數(shù)據(jù)。而對(duì)于擴(kuò)展區(qū)的映射過(guò)程，由于僅當(dāng)所有Mapper都計(jì)算完成才能獲得精確的擴(kuò)展區(qū)數(shù)據(jù)分布，因此若算法過(guò)早執(zhí)行，計(jì)數(shù)器counter的統(tǒng)計(jì)結(jié)果不夠精確，影響分區(qū)映射的合理性，而過(guò)晚執(zhí)行則會(huì)使Reducer處于饑餓狀態(tài)，影響了作業(yè)的執(zhí)行效率，因此分區(qū)映射算法的執(zhí)行時(shí)機(jī)應(yīng)設(shè)定為不影響作業(yè)執(zhí)行效率的最晚時(shí)間，即當(dāng)任意1個(gè)Reducer完成原生區(qū)的拉取工作，即啟動(dòng)第1次分區(qū)映射算法，而后續(xù)輪次分配的執(zhí)行時(shí)機(jī)均為上一輪拉取工作結(jié)束時(shí)間，以此類(lèi)推，完成整個(gè)擴(kuò)展區(qū)的映射過(guò)程。因此，每一輪分區(qū)映射過(guò)程都是對(duì)上一輪因統(tǒng)計(jì)結(jié)果不精確而產(chǎn)生的分配誤差進(jìn)行修正，從而經(jīng)過(guò)多輪迭代求得數(shù)據(jù)分配的近似最優(yōu)解。

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

本章通過(guò)實(shí)驗(yàn)比較和評(píng)價(jià)，驗(yàn)證迭代填充分區(qū)映射算法的有效性。

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

實(shí)驗(yàn)環(huán)境搭建采用1臺(tái)服務(wù)器和8個(gè)工作節(jié)點(diǎn)組成的集群，其中服務(wù)器作為Hadoop的NameNode和Spark的Master，主要配置為16顆4核心處理器陣列、256GB內(nèi)存和4個(gè)千兆網(wǎng)卡。8個(gè)工作節(jié)點(diǎn)作為DataNode和Slave，配置如表1所示。參數(shù)配置方面，HDFS的默認(rèn)備份數(shù)為3，Block大小為64MB，Spark的并行參數(shù)值(spark.default.parallelism)設(shè)置為16。作業(yè)執(zhí)行時(shí)間的監(jiān)測(cè)通過(guò)Spark控制臺(tái)，各種資源的使用狀況數(shù)據(jù)來(lái)源于nmon。

實(shí)驗(yàn)數(shù)據(jù)選取Zipf數(shù)據(jù)集和有向圖兩種類(lèi)型，其中Zipf數(shù)據(jù)集主要包括9個(gè)子數(shù)據(jù)集，總量為7.3GB，用于執(zhí)行WordCount作業(yè)。每個(gè)子數(shù)據(jù)集滿(mǎn)足指數(shù)為γ的標(biāo)準(zhǔn)Zipf分布，γ取值范圍為0.2～1.0的小數(shù)，增量為0.1，γ的取值越大，表示數(shù)據(jù)分布越傾斜。有向圖主要包括SNAP(StanfordNetworkAnalysisProject)[22]提供的標(biāo)準(zhǔn)數(shù)據(jù)集，用于執(zhí)行PageRank作業(yè)，如表2所示。

表1 工作節(jié)點(diǎn)配置參數(shù)

表2 測(cè)試數(shù)據(jù)集列表

4.2 擴(kuò)展參數(shù)評(píng)估實(shí)驗(yàn)

迭代填充分區(qū)映射算法通過(guò)引入擴(kuò)展參數(shù)，確定原生區(qū)與擴(kuò)展區(qū)的劃分比例，同時(shí)擴(kuò)展參數(shù)也決定了數(shù)據(jù)分配的輪數(shù)，因此實(shí)驗(yàn)首先驗(yàn)證擴(kuò)展參數(shù)對(duì)作業(yè)執(zhí)行效率的影響。實(shí)驗(yàn)選取Zipf數(shù)據(jù)集中γ取值為0.3、0.6和0.9的3個(gè)子數(shù)據(jù)集執(zhí)行WordCount作業(yè)，實(shí)驗(yàn)結(jié)果如圖2所示。

圖2 擴(kuò)展參數(shù)影響實(shí)驗(yàn)

由圖2可以看出，對(duì)于Zipf-0.3數(shù)據(jù)集，由于數(shù)據(jù)分布的傾斜度較低，其作業(yè)執(zhí)行效率隨擴(kuò)展參數(shù)值的增大，優(yōu)化效果并不明顯。而對(duì)于Zipf-0.6和Zipf-0.9，在前4個(gè)監(jiān)測(cè)點(diǎn)，隨著擴(kuò)展參數(shù)值的增大，作業(yè)執(zhí)行時(shí)間急劇下降，這是因?yàn)樵跀?shù)據(jù)分布傾斜度較大的情況下，原生區(qū)中各Bucket數(shù)據(jù)量差異較大，通過(guò)擴(kuò)展參數(shù)的介入，能夠附加額外的數(shù)據(jù)分配，修正原生區(qū)數(shù)據(jù)分配產(chǎn)生的誤差，因此擴(kuò)展參數(shù)值越大，修正效果越明顯。當(dāng)擴(kuò)展參數(shù)值為4時(shí)，作業(yè)執(zhí)行效率的優(yōu)化效果趨于穩(wěn)定，在后2個(gè)監(jiān)測(cè)點(diǎn)，作業(yè)執(zhí)行時(shí)間又出現(xiàn)小幅提高，這是由于擴(kuò)展系數(shù)具有最優(yōu)上限，在此基礎(chǔ)上繼續(xù)增加分配輪數(shù)也無(wú)法提高作業(yè)執(zhí)行效率；達(dá)到最優(yōu)上限后，算法的額外開(kāi)銷(xiāo)開(kāi)始顯現(xiàn)，額外開(kāi)銷(xiāo)導(dǎo)致了作業(yè)執(zhí)行效率的輕微下降。

4.3WordCount對(duì)比實(shí)驗(yàn)

實(shí)驗(yàn)選取5個(gè)不同分布的Zipf數(shù)據(jù)集執(zhí)行WordCount作業(yè)，對(duì)比迭代填充分區(qū)映射算法與傳統(tǒng)Spark的性能差異。其中擴(kuò)展參數(shù)值統(tǒng)一設(shè)置為4，Spark啟動(dòng)的Reducer數(shù)量為16。實(shí)驗(yàn)首先監(jiān)測(cè)最大負(fù)載節(jié)點(diǎn)和最小負(fù)載節(jié)點(diǎn)的分配數(shù)據(jù)量變化，實(shí)驗(yàn)結(jié)果如圖3所示。

由圖3可以看出，與傳統(tǒng)Spark環(huán)境相對(duì)比，迭代填充分區(qū)映射算法降低了最大負(fù)載工作節(jié)點(diǎn)的計(jì)算數(shù)據(jù)量，提高了最小負(fù)載節(jié)點(diǎn)的數(shù)據(jù)量。這是因?yàn)閭鹘y(tǒng)Spark一次分區(qū)策略對(duì)原始數(shù)據(jù)的分布不敏感，也缺乏有效的應(yīng)對(duì)策略，因此數(shù)據(jù)分布的傾斜性導(dǎo)致了最大、最小負(fù)載節(jié)點(diǎn)之間的數(shù)據(jù)分配差。而對(duì)于迭代填充分區(qū)算法，擴(kuò)展區(qū)的分配是對(duì)原生區(qū)數(shù)據(jù)傾斜分配有效彌補(bǔ)，從而產(chǎn)生相對(duì)均衡的數(shù)據(jù)分配。綜合來(lái)看，隨著Zipf分布指數(shù)的增大，傳統(tǒng)Spark的數(shù)據(jù)分配量差異越來(lái)越明顯，而迭代填充分區(qū)映射算法始終保持較為穩(wěn)定的均勻狀態(tài)。這是由于在傳統(tǒng)Spark環(huán)境下，數(shù)據(jù)傾斜度越大，工作節(jié)點(diǎn)數(shù)據(jù)量差異也越大，分配效果也越差。而迭代填充分區(qū)映射算法的映射過(guò)程是通過(guò)多輪次分配完成，每一輪分配都是對(duì)上一輪分配誤差的修正，有效降低了數(shù)據(jù)分布對(duì)分配效果的影響，因此，從數(shù)據(jù)分配合理性的角度來(lái)看，迭代填充分區(qū)映射算法具有良好的優(yōu)化效果。

圖3 分配數(shù)據(jù)量對(duì)比

圖4顯示了不同分布數(shù)據(jù)集作業(yè)執(zhí)行時(shí)間的對(duì)比，從實(shí)驗(yàn)結(jié)果來(lái)看，對(duì)于不同分布的Zipf數(shù)據(jù)集，迭代填充分區(qū)映射算法的作業(yè)執(zhí)行時(shí)間均小于傳統(tǒng)Spark的執(zhí)行時(shí)間。根據(jù)最大、最小節(jié)點(diǎn)的數(shù)據(jù)分配量可知，迭代填充分區(qū)映射算法保障了同構(gòu)集群環(huán)境上數(shù)據(jù)分配的均衡性，具有相同計(jì)算能力的Reducer分配計(jì)算量差異較小，各任務(wù)完成時(shí)間也較為接近，因此寬依賴(lài)Stage的執(zhí)行時(shí)間較短，作業(yè)的執(zhí)行效率更高。而對(duì)于傳統(tǒng)Spark，由于其對(duì)數(shù)據(jù)傾斜分布無(wú)任何有效應(yīng)對(duì)策略，往往導(dǎo)致相同計(jì)算能力Reducer所分配的計(jì)算數(shù)據(jù)量有較大差異，各任務(wù)完成時(shí)間長(zhǎng)短不一，因此寬依賴(lài)Stage的執(zhí)行時(shí)間較長(zhǎng)，降低了作業(yè)的整體執(zhí)行效率。從優(yōu)化效果來(lái)看，數(shù)據(jù)分布傾斜度越大，迭代填充分區(qū)映射算法的優(yōu)化效果越明顯，但優(yōu)化效果并未隨分布指數(shù)據(jù)的增大呈線(xiàn)性增加趨勢(shì)，這是因?yàn)闊o(wú)法預(yù)知精確的數(shù)據(jù)分布，迭代填充分區(qū)映射算法僅是在現(xiàn)有條件下提供較為均衡的數(shù)據(jù)分配方案，而且擴(kuò)展區(qū)的預(yù)留也采用固定的公式計(jì)算，不能根據(jù)不同數(shù)據(jù)分布進(jìn)行靈活的適應(yīng)性調(diào)整，因此對(duì)于不同的數(shù)據(jù)集其優(yōu)化效果無(wú)明顯規(guī)律。

圖4 WordCount作業(yè)執(zhí)行時(shí)間對(duì)比

4.4PageRank對(duì)比實(shí)驗(yàn)

上一節(jié)通過(guò)WordCount作業(yè)驗(yàn)證了算法的有效性，但由于WordCount僅包含一個(gè)依賴(lài)操作，Reducer也僅是作簡(jiǎn)單的加法運(yùn)算，不能完全體現(xiàn)迭代填充分區(qū)映射算法的優(yōu)化效果，因此本節(jié)選擇寬依賴(lài)個(gè)數(shù)更多、操作復(fù)雜度更高的PageRank作業(yè)對(duì)算法作進(jìn)一步評(píng)估。實(shí)驗(yàn)選擇了2個(gè)不同大小的數(shù)據(jù)集進(jìn)行，擴(kuò)展參數(shù)值為4，Reducer個(gè)數(shù)為16，實(shí)驗(yàn)結(jié)果如圖5所示。

圖5 PageRank作業(yè)執(zhí)行時(shí)間對(duì)比

由圖5可以看出，對(duì)于每一個(gè)數(shù)據(jù)集，傳統(tǒng)Spark和迭代填充分區(qū)映射算法的作業(yè)執(zhí)行時(shí)間都隨迭代次數(shù)的增加而上升；在每一個(gè)監(jiān)測(cè)點(diǎn)，傳統(tǒng)Spark的作業(yè)執(zhí)行時(shí)間均大于迭代填充分區(qū)映射算法，從而證明了本文算法對(duì)Spark框架的性能優(yōu)化具有良好的效果，也驗(yàn)證了理論模型及算法設(shè)計(jì)的正確性。從不同迭代次數(shù)的效率差異來(lái)看，作業(yè)的迭代次數(shù)越多，執(zhí)行時(shí)間的優(yōu)化度越高，作業(yè)執(zhí)行時(shí)間的縮減比基本隨迭代次數(shù)據(jù)增加呈線(xiàn)性增長(zhǎng)趨勢(shì)，這是由于在PageRank作業(yè)中，每輪迭代的數(shù)據(jù)分布都相同，因此迭代填充分區(qū)映射算法每輪迭代的優(yōu)化效果也基本相同。從作業(yè)執(zhí)行的整體趨勢(shì)來(lái)看，隨著迭代次數(shù)的增加，傳統(tǒng)Spark的作業(yè)執(zhí)行時(shí)間上升幅度較大，而迭代填充分區(qū)映射算法由于其多輪分配均衡了不同Reducer間的計(jì)算數(shù)據(jù)量，加速了寬依賴(lài)Stage的執(zhí)行，因此作業(yè)執(zhí)行時(shí)間上升趨勢(shì)較為緩和。由此可以看出，傳統(tǒng)Spark的作業(yè)執(zhí)行效率受寬依賴(lài)的影響較大，而迭代填充分區(qū)映射算法對(duì)寬依賴(lài)的敏感度較低，寬依賴(lài)Stage越多，越能夠體現(xiàn)算法的優(yōu)化效果，作業(yè)執(zhí)行的加速效應(yīng)也越明顯。

5 結(jié)語(yǔ)

本文針對(duì)Spark寬依賴(lài)Stage數(shù)據(jù)分區(qū)的傾斜問(wèn)題，首先分析Spark作業(yè)執(zhí)行機(jī)制，建立了作業(yè)效率模型，給出了RDD計(jì)算代價(jià)和作業(yè)執(zhí)行時(shí)間的定義。通過(guò)對(duì)Spark框架原始的分區(qū)策略進(jìn)行研究和分析，建立了分區(qū)映射模型，給出了源數(shù)據(jù)分布、分區(qū)映射和分配傾斜度的定義，并證明了這些定義對(duì)作業(yè)執(zhí)行效率影響，為算法設(shè)計(jì)提供依據(jù)。其次，在相關(guān)定義和證明的基礎(chǔ)上，提出擴(kuò)展式數(shù)據(jù)分區(qū)算法和迭代式分區(qū)映射算法，并對(duì)算法的執(zhí)行細(xì)節(jié)進(jìn)行分析和說(shuō)明。最后，通過(guò)不同的實(shí)驗(yàn)驗(yàn)證算法的有效性，實(shí)驗(yàn)結(jié)果表明，迭代填充分區(qū)映射算法提高了數(shù)據(jù)分配的合理性，優(yōu)化了寬依賴(lài)Stage的作業(yè)執(zhí)行效率。下一步的研究方向是探索異構(gòu)集群下適應(yīng)工作節(jié)點(diǎn)計(jì)算能力的分區(qū)映射策略。

)

[1]STRANDESM,CICOTTIP,SINKOVITSRS,etal.Gordon:design,performance,andexperiencesdeployingandsupportingadataintensivesupercomputer[C]//Proceedingsofthe1stConferenceoftheExtremeScienceandEngineeringDiscoveryEnvironment:BridgingfromtheExtremetotheCampusandBeyond.NewYork:ACM, 2012:ArticleNo. 3.

[2]BRONEVETSKYG,MOODYA.ScalableI/Osystemsvianode-localstorage:approaching1TB/secfileI/O,LLNL-TR-415791 [R].Livermore,CA:LawrenceLivermoreNationalLaboratory, 2009: 1-6.

[3]ZAHARIAM,CHOWDHURYM,DAST,etal.FastandinteractiveanalyticsoverHadoopdatawithSpark[J].Login, 2012, 37(4): 45-51.

[4]ApacheSpark.Sparkoverview[EB/OL]. [2015- 03- 18].http://spark.apache.org.

[5]ZAHARIAM,CHOWDHURYM,DAST,etal.Resilientdistributeddatasets:afault-tolerantabstractionforin-memoryclustercomputing[C]//Proceedingsofthe9thUSENIXConferenceonNetworkedSystemsDesignandImplementation.Berkeley,CA:USENIXAssociation, 2012: 2.

[6]LINX,WANGP,WUB.LoganalysisincloudcomputingenvironmentwithHadoopandSpark[C]//Proceedingsofthe5thIEEEInternationalConferenceonBroadbandNetworkandMultimediaTechnology.Piscataway,NJ:IEEE, 2013: 273-276.

[7]DONGX,XIEY,MURALIMANOHARN,etal.Hybridcheckpointingusingemergingnonvolatilememoriesforfutureexascalesystems[J].ACMTransactionsonArchitectureandCodeOptimization, 2011, 8(2): 510-521.

[8] 慈軼為,張展,左德承,等.可擴(kuò)展的多周期檢查點(diǎn)設(shè)置[J].軟件學(xué)報(bào),2010,21(2):218-230.(CIYW,ZHANGZ,ZUODC,etal.Scalabletime-basedmulti-cyclecheckpointing[J].JournalofSoftware, 2010, 21(2): 218-230.)

[9]DEANJ,GHEMAWATS.MapReduce:simplifieddataprocessingonlargeclusters[C]//Proceedingsofthe6thConferenceonSymposiumonOpeartingSystemsDesignandImplementation.Berkeley,CA:USENIXAssociation, 2004,6: 10.

[10]KWONY,BALAZINSKAM,HOWEB,etal.AstudyofskewinMapReduceapplication[EB/OL]. [2016- 03- 18].https://www.researchgate.net/publication/228941278_A_Study_of_Skew_in_MapReduce_Applications.

[11]KWONY,BALAZINSKAM,HOWEB,etal.Skew-resistantparallelprocessingoffeature-extractingscientificuser-definedfunctions[C]//Proceedingsofthe1stACMSymposiumonCloudComputing.NewYork:ACM, 2010: 75-86.

[12] 王卓,陳群,李戰(zhàn)懷,等.基于增量式分區(qū)策略的MapReduce數(shù)據(jù)均衡方法[J].計(jì)算機(jī)學(xué)報(bào),2016,39(1):19-35.(WANGZ,CHENQ,LIZH,etal.AnincrementalpartitioningstrategyfordatabalanceonMapReduce[J].ChineseJournalofComputers, 2016, 39(1): 19-35.)

[13]KWONY,BALAZINSKAM,HOWEB,etal.SkewTune:mitigatingskewinMapReduceapplications[C]//Proceedingsofthe2012ACMSIGMODInternationalConferenceonManagementofData.NewYork:ACM, 2012: 25-36.

[14]YANW,XUEY,MALINB.ScalableandrobustkeygroupsizeestimationforreducerloadbalancinginMapReduce[C]//Proceedingsofthe2013IEEEInternationalConferenceonBigData.Piscataway,NJ:IEEE, 2013: 156-162.

[15]RAMAKRISHNANSR,SWARTG,URMANOVA,etal.BalancingreducerskewinMapReduceworkloadsusingprogressivesampling[C]//Proceedingsofthe3rdACMSymposiumonCloudComputing.NewYork:ACM, 2012:ArticleNo. 16.

[16]GUFLERB,AUGSTENN,REISERA,etal.HandingdataskewinMapReduce[C]//Proceedingsofthe1stInternationalConferenceonCloudComputingandServicesScience.Berlin:Springer, 2011: 574-583.

[17]GUFLERB,AUGSTENN,REISERA,etal.LoadbalancinginMapReducebasedonscalablecardinalityestimates[C]//Proceedingsofthe2012IEEE28thInternationalConferenceonDataEngineering.Washington,DC:IEEEComputerSociety, 2012: 522-533.

[18]KOLBL,THORA,RAHME.LoadbalancingforMapReduce-basedentityresolution[C]//Proceedingsofthe2012IEEE28thInternationalConferenceonDataEngineering.Washington,DC:IEEEComputerSociety, 2012: 618-629.

[19]KOLBL,THORA,RAHME,etal.Block-basedloadbalancingforentityresolutionwithMapReduce[C]//Proceedingsofthe20thACMInternationalConferenceonInformationandKnowledgeManagement.NewYork:ACM, 2011: 2397-2400.

[20]RACHASC.LoadbalancingMap-Reducecommunicationsforefficientexecutionsofapplicationsinacloud[D].Bangalore,India:IndianInstituteofScience, 2012: 12-16.

[21]IBRAHIMS,JINH,LUL,etal.HandlingpartitioningskewinMapReduceusingLEEN[J].Peer-to-PeerNetworkingandApplications, 2013, 6(4): 409-424.

[22]JUREL.Stanfordnetworkanalysisproject[EB/OL]. [2015- 03- 18].http://snap.stanford.edu.

ThisworkispartiallysupportedbytheNationalNaturalScienceFoundationofChina(61262088, 61462079, 61363083, 61562086),theEducationalResearchProgramofXinjiangUygurAutonomousRegion(XJEDU2016S106).

BIAN Chen, born in 1981, Ph. D. candidate, associate professor. His research interests include parallel computing, distributed system.

YU Jiong, born in 1964, Ph. D., professor. His research interests include grid computing, high performance computing.

XIU Weirong, born in 1979, M. S., lecturer. Her research interests include data mining, distributed applications.

YING Changtian, born in 1989. Ph. D. candidate. Her research interests include big data storage, in-memory computing.

Qian Yurong, born in 1980. Ph. D., associate professor. Her research interests include cloud computing, graphics and image processing.

Partitioning and mapping algorithm for in-memory computing framework based on iterative filling

BIAN Chen*, YU Jiong, XIU Weirong, YING Changtian, QIAN Yurong

(SchoolofInformationScienceandEngineering,XinjiangUniversity,UrumqiXinjiang830046,China)

Focusing on the issue that the only one Hash/Range partitioning strategy in Spark usually results in unbalanced data load at Reduce phase and increases job duration sharply, an Iterative Filling data Partitioning and Mapping algorithm (IFPM) which include several innovative approaches was proposed. First of all, according to the analysis of job execute scheme of Spark, the job efficiency model and partition mapping model were established, the definitions of job execute timespan and allocation incline degree were given. Moreover, the Extendible Partitioning Algorithm (EPA) and Iterative Mapping Algorithm (IMA) were proposed, which reserved partial data into extend region by one-to-many partition function at Map phase. Data in extended region would be mapped by extra iterative allocation until the approximate data distribution was obtained, and the adaptive mapping function was executed by awareness of calculated data size at Reduce phase to revise the unbalanced data load in original region allocation. Experimental results demonstrate that for any distribution of the data, IFPM promotes the rationality of data load allocation from Map phase to Reduce phase and optimize the job efficiency of in-memory computing framework.

in-memory computing; load balance; extendible partitioning; iterative mapping

2016- 09- 26;

2016- 10- 17。

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

卞琛(1981—)，男，江蘇南京人，副教授，博士研究生，CCF會(huì)員，主要研究方向：網(wǎng)絡(luò)計(jì)算、分布式系統(tǒng)； 于炯(1964—)，男，北京人，教授，博士，CCF高級(jí)會(huì)員，主要研究方向：網(wǎng)格計(jì)算、高性能計(jì)算； 修位蓉(1979—)，女，重慶人，講師，碩士，主要研究方向：數(shù)據(jù)挖掘、分布式應(yīng)用； 英昌甜(1989—)，女，新疆烏魯木齊人，博士研究生，主要研究方向：大數(shù)據(jù)存儲(chǔ)、內(nèi)存計(jì)算； 錢(qián)育蓉(1979—)，女，新疆烏魯木齊人，副教授，博士，CCF會(huì)員，主要研究方向：云計(jì)算、圖形圖像處理。

1001- 9081(2017)03- 0647- 07

10.11772/j.issn.1001- 9081.2017.03.647

TP

A