廖 彬，張 陶，國(guó)冰磊，于 炯，張旭光，劉 炎

(1.新疆財(cái)經(jīng)大學(xué) 統(tǒng)計(jì)與信息學(xué)院，烏魯木齊 830012； 2.新疆醫(yī)科大學(xué) 醫(yī)學(xué)工程技術(shù)學(xué)院，烏魯木齊 830011；3.新疆大學(xué) 信息科學(xué)與工程學(xué)院，烏魯木齊 830008； 4.清華大學(xué) 軟件學(xué)院，北京 100084) (*通信作者電子郵箱liaobin665@163.com)

基于Spark的ItemBased推薦算法性能優(yōu)化

廖 彬1*，張 陶2,3，國(guó)冰磊3，于 炯3，張旭光1，劉 炎4

(1.新疆財(cái)經(jīng)大學(xué) 統(tǒng)計(jì)與信息學(xué)院，烏魯木齊 830012； 2.新疆醫(yī)科大學(xué) 醫(yī)學(xué)工程技術(shù)學(xué)院，烏魯木齊 830011；3.新疆大學(xué) 信息科學(xué)與工程學(xué)院，烏魯木齊 830008； 4.清華大學(xué) 軟件學(xué)院，北京 100084) (*通信作者電子郵箱liaobin665@163.com)

MapReduce計(jì)算場(chǎng)景下，復(fù)雜的大數(shù)據(jù)挖掘類算法通常需要多個(gè)MapReduce作業(yè)協(xié)作完成，但多個(gè)作業(yè)之間嚴(yán)重的冗余磁盤讀寫及重復(fù)的資源申請(qǐng)操作，使得算法的性能嚴(yán)重降低。為提高ItemBased推薦算法的計(jì)算效率，首先對(duì)MapReduce平臺(tái)下ItemBased協(xié)同過(guò)濾算法存在的性能問題進(jìn)行了分析；在此基礎(chǔ)上利用Spark迭代計(jì)算及內(nèi)存計(jì)算上的優(yōu)勢(shì)提高算法的執(zhí)行效率，并實(shí)現(xiàn)了基于Spark平臺(tái)的ItemBased推薦算法。實(shí)驗(yàn)結(jié)果表明：當(dāng)集群節(jié)點(diǎn)規(guī)模分別為10與20時(shí)，算法在Spark中的運(yùn)行時(shí)間分別只有MapReduce中的25.6%及30.8%，Spark平臺(tái)下的算法相比MapReduce平臺(tái)，執(zhí)行效率整體提高3倍以上。

協(xié)同過(guò)濾；MapReduce；Spark算法；性能優(yōu)化；有向非循環(huán)圖

0 引言

據(jù)互聯(lián)網(wǎng)數(shù)據(jù)中心(Internet Data Center)發(fā)布的報(bào)告顯示，2015年全球產(chǎn)生的數(shù)據(jù)量達(dá)到近10 ZB，而2020年全球產(chǎn)生的數(shù)據(jù)量將達(dá)到40 ZB[1]。數(shù)據(jù)的產(chǎn)生過(guò)程在經(jīng)歷被動(dòng)和主動(dòng)兩種產(chǎn)生過(guò)程后，發(fā)展到了自動(dòng)產(chǎn)生階段，標(biāo)志著大數(shù)據(jù)時(shí)代的來(lái)臨。數(shù)據(jù)從簡(jiǎn)單的處理對(duì)象開始轉(zhuǎn)變?yōu)橐环N基礎(chǔ)性資源，如何更好地管理和利用大數(shù)據(jù)已經(jīng)成為普遍關(guān)注的話題，大數(shù)據(jù)的規(guī)模效應(yīng)給數(shù)據(jù)存儲(chǔ)、管理以及數(shù)據(jù)分析帶來(lái)了極大的挑戰(zhàn)[2]，進(jìn)而高效率低成本的大數(shù)據(jù)處理技術(shù)成為學(xué)術(shù)界及工業(yè)界的研究熱點(diǎn)。推薦算法作為大數(shù)據(jù)挖掘類算法中最常見的一種，由于可以快速地在海量信息中篩選并過(guò)濾出用戶可能最感興趣的內(nèi)容，在幫助用戶獲取有效信息的同時(shí)實(shí)現(xiàn)信息提供商與用戶之間的雙贏，得到了快速的發(fā)展，而協(xié)同過(guò)濾推薦(Collaborative Filtering Recommendation)是其中最經(jīng)典的推薦算法。

自2003年Google發(fā)表論文公開分布式存儲(chǔ)系統(tǒng)GFS(Google File System)[3]及分布式數(shù)據(jù)處理模型MapReduce[4-5]以來(lái)，由于MapReduce具有可擴(kuò)展性(Scalable)、低成本(Economical)、高效性(Efficient)及可靠性(Reliable)等優(yōu)點(diǎn)[6]，使其成為諸多大數(shù)據(jù)計(jì)算系統(tǒng)(如：Hadoop、Spark、Pig、Hive、Hbase和Dryad等)最通用的并行計(jì)算框架(MapReduce生態(tài)系統(tǒng)如圖1[7]所示)。MapReduce與以往的并行計(jì)算模式相比，最大的區(qū)別是它“移動(dòng)計(jì)算”而非“移動(dòng)數(shù)據(jù)”的理念，即將任務(wù)調(diào)度到離數(shù)據(jù)最近的節(jié)點(diǎn)，因?yàn)檫@樣可以最小化大數(shù)據(jù)作業(yè)計(jì)算過(guò)程中數(shù)據(jù)的網(wǎng)絡(luò)傳輸量，而這正是MapReduce發(fā)展如此壯大的根本原因。

圖1 MapReduce生態(tài)系統(tǒng)[7]

在此背景下，隨著MapReduce計(jì)算框架的不斷發(fā)展壯大，已有的推薦算法大多基于MapReduce框架實(shí)現(xiàn)，但MapReduce計(jì)算場(chǎng)景下，復(fù)雜的大數(shù)據(jù)挖掘類算法通常需要多個(gè)MapReduce作業(yè)協(xié)作完成，但多個(gè)作業(yè)之間嚴(yán)重的冗余磁盤讀寫及重復(fù)的資源申請(qǐng)操作，使得算法的計(jì)算及能耗等效率嚴(yán)重降低[8-10]。而協(xié)同過(guò)濾推薦算法則屬于典型的大數(shù)據(jù)挖掘類算法，所以本文以基于物品(ItemBased)的協(xié)同過(guò)濾算法為例，研究提高算法性能的優(yōu)化方法。本文作了如下幾個(gè)方面的工作：首先，對(duì)基于MapReduce的ItemBased協(xié)同過(guò)濾算法實(shí)現(xiàn)及性能缺陷進(jìn)行了分析；其次，在性能缺陷分析的基礎(chǔ)上，提出基于Spark平臺(tái)重寫算法，進(jìn)而提高ItemBased推薦算法的性能；最后，通過(guò)實(shí)驗(yàn)驗(yàn)證了算法優(yōu)化效果的有效性。

1 相關(guān)研究

由于應(yīng)用領(lǐng)域及場(chǎng)景的不同，最常見的協(xié)同過(guò)濾算法[11]有基于用戶、物品及混合協(xié)同過(guò)濾3種，除此之外還有基于內(nèi)容的協(xié)同過(guò)濾[12-13]、基于規(guī)則的協(xié)同過(guò)濾[14-15]、基于人口統(tǒng)計(jì)信息的協(xié)同過(guò)濾[16-17]、基于網(wǎng)絡(luò)結(jié)構(gòu)的協(xié)同過(guò)濾[18-19]及混合過(guò)濾[20]等。已有的推薦算法研究工作主要目的都是以提高算法在特定應(yīng)用領(lǐng)域的推薦質(zhì)量為首要目標(biāo)，主要評(píng)價(jià)指標(biāo)有準(zhǔn)確率(Precision)、召回率(Recall)、F值(F-Measure)、E值(E-Measure)、平均正確率(Average Precision)等。很少工作關(guān)注到算法的計(jì)算效率、算法執(zhí)行環(huán)境等方面，但在算法處理數(shù)據(jù)量的爆炸式增長(zhǎng)及分布式計(jì)算模型MapReduce逐漸普及的背景下，推薦算法從單機(jī)模型逐漸向MapReduce發(fā)展，并且算法的執(zhí)行效率也逐漸受到學(xué)術(shù)界及工業(yè)界的重視。

文獻(xiàn)[21-22]將基于用戶(User-Based)的協(xié)同過(guò)濾算法移植到Hadoop平臺(tái)，以提高算法并行計(jì)算的能力，并通過(guò)修改作業(yè)map及reduce子任務(wù)的數(shù)量，達(dá)到提高M(jìn)apReduce模型下算法計(jì)算效率的目的。Schelter等[23]面對(duì)用戶數(shù)據(jù)量快速增長(zhǎng)的問題，開發(fā)了基于MapReduce的近鄰(Similarity-based neighborhood)協(xié)同過(guò)濾算法，并通過(guò)7億Yahoo音樂數(shù)據(jù)的實(shí)驗(yàn)證明了算法在效率上的提升。文獻(xiàn)[24]將MinHash聚類、概率潛在語(yǔ)義索引(Probabilistic Latent Semantic Indexing, PLSI)及Covisitation計(jì)數(shù)技術(shù)引入到Google news的協(xié)同過(guò)濾算法中，有效地提高了算法的性能。文獻(xiàn)[25]開發(fā)并實(shí)現(xiàn)了MapReduce平臺(tái)下基于物品(ItemBased)協(xié)同過(guò)濾算法，將算法的計(jì)算步驟切分為Map及Reduce子任務(wù)，并通過(guò)數(shù)據(jù)本地化策略最小化通信成本，大大提高了Hadoop中算法的執(zhí)行效率。但是，不管是UserBased還是ItemBased協(xié)同過(guò)濾算法，MapReduce環(huán)境下的計(jì)算任務(wù)都需要多個(gè)作業(yè)協(xié)作完成，作業(yè)之間難免存在冗余的I/O及資源重復(fù)申請(qǐng)操作，算法效率上還存在著較大的優(yōu)化空間。所以，本文與以往工作不同的是：在分析基于MapReduce的ItemBase協(xié)同過(guò)濾算法性能缺陷的基礎(chǔ)上，利用Spark在迭代計(jì)算和內(nèi)存計(jì)算上的優(yōu)勢(shì)，提高算法的執(zhí)行效率。

2 基于MapReduce的ItemBased推薦算法分析

選擇ItemBased的協(xié)同過(guò)濾算法為研究對(duì)象主要原因是：為了讓相似度矩陣計(jì)算規(guī)模最小化，ItemBased算法更加適用于用戶數(shù)據(jù)量比物品數(shù)據(jù)量大的場(chǎng)景，而UserBased則適用于用戶數(shù)比物品數(shù)小的場(chǎng)景。除此之外，相對(duì)于其他協(xié)同過(guò)濾算法，item與item之間的相似度比較穩(wěn)定，適合離線計(jì)算，能夠?qū)崿F(xiàn)定期更新的功能，這使得ItemBased協(xié)同過(guò)濾算法的應(yīng)用場(chǎng)景相比其他推薦算法更為廣泛。

ItemBased算法假設(shè)能夠引起用戶興趣的item必定與其評(píng)分高的item相似。算法首先計(jì)算用戶對(duì)物品的喜好程度，然后根據(jù)用戶的喜好計(jì)算item之間的相似度，最后找出與每個(gè)item最相似的前N個(gè)item。其中，相似性計(jì)算包含3個(gè)步驟:1)統(tǒng)計(jì)每個(gè)item的好評(píng)用戶數(shù);2)item好評(píng)鍵值對(duì)統(tǒng)計(jì)，即統(tǒng)計(jì)任意兩個(gè)有關(guān)聯(lián)item的相同好評(píng)用戶數(shù);3)item相似性計(jì)算(可采用Jaccard系數(shù)作為計(jì)算兩個(gè)item的相似性方法)，即計(jì)算任意兩個(gè)有關(guān)聯(lián)item的相似度。而找出最相似的前N個(gè)item步驟中，首先需要對(duì)item相似性歸一化，再進(jìn)行相似性評(píng)分整合，最后得出每個(gè)item相似性最高的前N個(gè)item。MapReduce的ItemBased推薦算法，其最經(jīng)典的實(shí)現(xiàn)代碼是Mahout中的RecommenderJob類，其源碼在org.apache.mahout.cf.taste.hadoop.item包下，其主要參數(shù)及意義如表1所示。

表1 ItemBased推薦算法主要參數(shù)

通過(guò)對(duì)ItemBased推薦算法在Hadoop中的運(yùn)行過(guò)程監(jiān)控及日志分析，發(fā)現(xiàn)執(zhí)行一次算法總共包含12個(gè)MapReduce作業(yè)，根據(jù)不同MapReduce作業(yè)的功能，可分解為如下10個(gè)計(jì)算步驟，每個(gè)MapReduce作業(yè)所對(duì)應(yīng)的計(jì)算步驟及功能如表2。

表2 算法MapReduce作業(yè)分解及其功能

即使將MapReduce Job1(MR Job1)、MapReduce Job11等這樣的非核心的計(jì)算步驟去掉，最精簡(jiǎn)的ItemBased推薦算法也至少需要7個(gè)MapReduce作業(yè)才能完成計(jì)算。如圖2所示為精簡(jiǎn)后的MapReduce作業(yè)與HDFS(Hadoop Distributed File System)文件系統(tǒng)的調(diào)用關(guān)系。

根據(jù)MapReduce計(jì)算模型的原理，每個(gè)MapReduce作業(yè)會(huì)被切分為多個(gè)map及reduce任務(wù)。其中map任務(wù)執(zhí)行過(guò)程中需要將數(shù)據(jù)從HDFS中讀取進(jìn)來(lái)，當(dāng)map任務(wù)計(jì)算完畢后，通過(guò)Shuffle與Sort操作將〈key,value〉鍵值對(duì)發(fā)送給對(duì)應(yīng)的reduce任務(wù)，reduce階段以〈key,Iterator〈value〉〉作為數(shù)據(jù)輸入，計(jì)算完畢后將處理結(jié)果寫入到HDFS中。由于MapReduce作業(yè)之間相互獨(dú)立，即每個(gè)MapReduce作業(yè)都要進(jìn)行輸入數(shù)據(jù)的讀取及將計(jì)算結(jié)果寫入HDFS的操作。由于ItemBased推薦算法總共包含12個(gè)MapReduce作業(yè)，意味著算法執(zhí)行過(guò)程中需要進(jìn)行總共24次的HDFS讀取及寫入操作。這些重復(fù)的數(shù)據(jù)讀寫操作耗費(fèi)集群資源的同時(shí)，嚴(yán)重降低了算法的執(zhí)行的效率；并且，由于磁盤I/O資源是集群計(jì)算性能的瓶頸，高頻次的I/O資源申請(qǐng)及釋放容易產(chǎn)生資源競(jìng)爭(zhēng)，導(dǎo)致map與reduce任務(wù)之間容易出現(xiàn)等待現(xiàn)象，進(jìn)而不同程度上進(jìn)一步降低了算法的執(zhí)行效率。

通過(guò)以上分析發(fā)現(xiàn)MapReduce環(huán)境下的ItemBased算法性能優(yōu)化并不能單從優(yōu)化算法本身入手，MapReduce作業(yè)之間的獨(dú)立性所帶來(lái)的重復(fù)I/O操作是導(dǎo)致算法的計(jì)算效率低下的根本原因。

圖2 MapReduce作業(yè)與HDFS的調(diào)用關(guān)系

3 基于Spark的ItemBased推薦算法

第2章分析了ItemBased推薦算法在MapReduce平臺(tái)下效率低下是由于MapReduce作業(yè)之間的獨(dú)立性及資源的不合理利用造成，Spark相比MapReduce主要有以下兩方面的優(yōu)勢(shì)。

1)Spark能夠?qū)⒂?jì)算過(guò)程中產(chǎn)生的中間數(shù)據(jù)緩存到內(nèi)存，與MapReduce相比，迭代計(jì)算效率更高。這是由于Spark的RDD(Resilient Distributed Dataset)能夠直接cache到內(nèi)存中。Action或Transformation算子對(duì)RDD數(shù)據(jù)集的計(jì)算結(jié)果緩存到內(nèi)存中，使得下一個(gè)算子的輸入數(shù)據(jù)直接從內(nèi)存中讀取，相比MapReduce省去了大量的重復(fù)磁盤I/O操作。

2)Spark實(shí)現(xiàn)推薦算法業(yè)務(wù)邏輯比MapReduce更簡(jiǎn)單靈活。這是由于MapReduce只提供map與reduce兩種基本操作，而Spark則提供包含Transfromation及Action兩類操作集。其中，Transfromation中包含map、filter、flatMap、sample、groupByKey、reduceByKey、union、join、cogroup、mapValues、sort和partionBy等操作；Action中擁有count、collect、reduce、lookup和save等操作。除此之外，Spark能夠?qū)⑦@些算子之間的業(yè)務(wù)邏輯(計(jì)算流程)通過(guò)有向非循環(huán)圖(Directed Acyclic Graph, DAG)模型進(jìn)行組織調(diào)度。

由于目前Spark MLlib中已有的推薦算法只有ALS(Alternating Least Squares)一種，并沒有ItemBased推薦算法。所以，本文對(duì)Spark平臺(tái)下基于Scala語(yǔ)言對(duì)算法進(jìn)行了實(shí)現(xiàn)。算法核心主要包含以下3個(gè)部分。

1)數(shù)據(jù)輸入模型構(gòu)建。

ItemBased推薦算法輸入數(shù)據(jù)格式為：(userID,itemID,ratingScore)意義分別為用戶ID、物品ID和評(píng)分值。構(gòu)造用戶數(shù)據(jù)輸入模型核心代碼如下：

def UserData ( sc:SparkContext,input:String,split:String):(RDD[(String,String,Double)])={ val u_rdd1=sc.textFile(input,10)

val u_rdd2=u_rdd1.map(line=>{ val fileds=line.split("split") (fileds(0),fileds(1),fileds(2).toDouble) }) u_rdd2

}

2)相似度矩陣模型構(gòu)建及計(jì)算。

可將相似度計(jì)算的輸入數(shù)據(jù)抽象的表示為兩個(gè)關(guān)系模式：Features(id,feature)和Relationship(id,fid)。其中Features表示每個(gè)對(duì)象(如用戶、文本、商品等)所具有的特征信息；而Relationship表示Features中的對(duì)象之間存在的關(guān)系。相似度計(jì)算的流程是：首先，通過(guò)Relationship表中的關(guān)系對(duì)(id,fid)，關(guān)聯(lián)到Features表中id及fid中所對(duì)應(yīng)的feature字段的值(即〈f1,f2,…,fn〉的值)；其次，將兩組feature的值作為輸入?yún)?shù)，交給相似度計(jì)算函數(shù)SC(Similarity Calculation)進(jìn)行計(jì)算。最后，將每次SC函數(shù)計(jì)算結(jié)果輸出到結(jié)果表中。可根據(jù)具體的應(yīng)用場(chǎng)景不同而選擇不同的相似度計(jì)算函數(shù)，常用的相似度計(jì)算函數(shù)SC有Cosine Correlation函數(shù)、Spearman Rank Correlation函數(shù)、Pearson Correlation函數(shù)及歐氏距離相似度函數(shù)等。本文采用歐氏相似度函數(shù)，其中輸入?yún)?shù)u_rdd表示用戶評(píng)分表，輸出參數(shù)u_rdd9表示相似矩陣，數(shù)據(jù)格式為：(itemID1,itemID2,相似度)。其核心代碼EuclideanDistanceSimilarity如下：

def EuclideanDistanceSimilarity( u_rdd:RDD[(String,String,Double)]):(RDD[(String,String,Double)])={ val u_rdd2=u_rdd.map(f=>(f._1,(f._2,f._3))).sortByKey()

u_rdd2.cache

val u_rdd3=u_rdd2.join(u_rdd2)

val u_rdd4=u_rdd3.map(f=>((f._2._1._1,f._2._2._1), (f._2._1._2,f._2._2._2))) val u_rdd5=u_rdd4.map(f=>(f._1,(f._2._1-f._2._2)* (f._2._1-f._2._2))).reduceByKey(_+_) val u_rdd6=u_rdd4.map(f=>(f._1,1)).reduceByKey(_+_)

val u_rdd7=u_rdd5.filter(f=>f._1._1!=f._1._2)

val u_rdd8=u_rdd7.join(u_rdd6)

val u_rdd9=u_rdd8.map(f=>(f._1._1,f._1._2,f._2._2/ (1+sqrt(f._2._1)))) u_rdd9

}

3)推薦算法模型構(gòu)建及業(yè)務(wù)實(shí)現(xiàn)。

在構(gòu)建用戶輸入模型及相似度計(jì)算模型的基礎(chǔ)上，推薦模型根據(jù)物品相似矩陣以及用戶對(duì)物品的評(píng)分表，計(jì)算用戶的推薦列表。其中，輸入?yún)?shù)items_similar表示物品相似矩陣，user_perf表示用戶評(píng)分表，r_number表示推薦個(gè)數(shù)；而輸出參數(shù)rdd7表示用戶推薦列表，格式為：(userID,itemID,ratingScore)。其核心代碼Recommend如下。

def Recommend ( items_similar:RDD[(String,String,Double)],user_perf:RDD[(String,String,Double)],r_number:Int):(RDD[(String,String,Double)])={ val rdd2=items_similar.map(f=>(f._2,(f._1,f._3))). join(user_perf.map(f=>(f._2,(f._1,f._3)))) val rdd3=rdd2.map(f=>((f._2._2._1,f._2._1._1), f._2._2._2*f._2._1._2)) val rdd4=rdd3.reduceByKey((x,y)=>x+y). map(f=>(f._1._1,(f._1._2,f._2))) val rdd5=rdd4.groupByKey()

val rdd6=rdd5.map(f=>{ val i2=f._2.toBuffer val i2_2=i2.sortBy(_._2) if (i2_2.length>r_number)i2_2.remove(0,(i2_2. length-r_number))(f._1,i2_2.toIterable) })

val rdd7=rdd6.flatMap(f=>{ val id2=f._2 for (w<-id2)yield(f._1,w._1,w._2) })

rdd7

}

算法在Spark環(huán)境中的計(jì)算流程，RDD數(shù)據(jù)集之間的轉(zhuǎn)化操作及關(guān)系，請(qǐng)參見圖3。

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

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

為了對(duì)比算法在MapReduce與Spark平臺(tái)下的執(zhí)行效率，本文利用測(cè)試數(shù)據(jù)生成工具DataFactory生成測(cè)試數(shù)據(jù)8億條。將測(cè)試數(shù)據(jù)生成到數(shù)據(jù)庫(kù)表中后，將數(shù)據(jù)庫(kù)表中的測(cè)試數(shù)據(jù)導(dǎo)出為TXT數(shù)據(jù)后上傳到HDFS中。HDFS中數(shù)據(jù)塊的大小設(shè)置為16 MB，數(shù)據(jù)的切分及存儲(chǔ)情況由機(jī)架感知的數(shù)據(jù)塊存儲(chǔ)策略隨機(jī)確定。測(cè)試數(shù)據(jù)的格式為(userid，itemid，ratingScore)，其含義表示用戶(userid)對(duì)物品(itemid)的評(píng)分結(jié)果為ratingScore。實(shí)驗(yàn)在2組不同規(guī)模集群上進(jìn)行測(cè)試，第1組節(jié)點(diǎn)規(guī)模為10，第2組節(jié)點(diǎn)規(guī)模為20。實(shí)驗(yàn)節(jié)點(diǎn)的配置參數(shù)如表3所示。

表3 總體實(shí)驗(yàn)環(huán)境描述

4.2 算法執(zhí)行效率對(duì)比

為了對(duì)比算法在MapReduce與Spark平臺(tái)下的執(zhí)行效率，在輸入數(shù)據(jù)相同的條件下，分別將MapReduce及Spark平臺(tái)下的算法在節(jié)點(diǎn)為10與20的集群中運(yùn)行10次，得到的實(shí)驗(yàn)數(shù)據(jù)如表4所示。其中：在MapReduce 10節(jié)點(diǎn)的集群環(huán)境下算法運(yùn)行10次的平均時(shí)間為13 226.7 s，在集群節(jié)點(diǎn)規(guī)模為20時(shí)平均時(shí)間縮短到7 117.4 s；Spark集群規(guī)模為10個(gè)節(jié)點(diǎn)時(shí)平均時(shí)間為3 380.5 s，當(dāng)Spark集群規(guī)模擴(kuò)大到20個(gè)節(jié)點(diǎn)時(shí)平均時(shí)間縮短至2 194.4 s。

從表3中數(shù)據(jù)可以看出，由于算法執(zhí)行的并行性，無(wú)論是MapReduce還是Spark平臺(tái)，隨著集群規(guī)模的擴(kuò)大，算法運(yùn)行時(shí)間有所縮短。其中MapReduce集群規(guī)模從10節(jié)點(diǎn)擴(kuò)充到20后，算法平均執(zhí)行時(shí)間縮短了6 109.3 s，效率提升了近46.2%；而Spark集群規(guī)模的擴(kuò)大，算法平均執(zhí)行時(shí)間縮短了1 186.1 s，效率提升了35.1%左右。當(dāng)集群規(guī)模都為10節(jié)點(diǎn)時(shí)，Spark整體上比MapReduce平均執(zhí)行時(shí)間縮短9 846.2 s，Spark算法運(yùn)行時(shí)間只有MapReduce的25.6%；而當(dāng)集群規(guī)模為20時(shí)，Spark整體上比MapReduce平均執(zhí)行時(shí)間縮短4 923 s，Spark算法運(yùn)行時(shí)間只有MapReduce的30.8%。

4.3 算法執(zhí)行效率分析

如圖3所示為算法在Spark環(huán)境中的計(jì)算流程分析，整個(gè)算法在Spark環(huán)境中運(yùn)行分為8個(gè)Stage，涉及到18個(gè)RDD數(shù)據(jù)集，圖3描述了RDD數(shù)據(jù)集之間的轉(zhuǎn)化關(guān)系。將圖3的計(jì)算過(guò)程與圖2進(jìn)行對(duì)比，可以看出相對(duì)于MapReduce平臺(tái)，Spark平臺(tái)下對(duì)算法的執(zhí)行時(shí)間及資源利用效率提升主要有以下3個(gè)方面的原因。

1)當(dāng)RDD緩存到內(nèi)存時(shí)，相比直接從HDFS中讀取數(shù)據(jù)，效率提高很多。當(dāng)算法在Spark平臺(tái)中執(zhí)行時(shí)，中間數(shù)據(jù)以RDD的方式緩存在內(nèi)存中，相對(duì)于MapReduce每個(gè)作業(yè)都需要進(jìn)行磁盤的讀寫操作，大大地提高了磁盤I/O資源的利用效率。RDD的Cache機(jī)制減少磁盤I/O壓力的同時(shí)，還能提高數(shù)據(jù)并行讀取的能力，比如RDD3進(jìn)行Cache后，RDD4和RDD7都可以同時(shí)訪問RDD3的數(shù)據(jù)，這比MapReduce作業(yè)之間重復(fù)從HDFS中讀取相同數(shù)據(jù)的問題,數(shù)據(jù)訪問效率再次提升。

表4 MapReduce與Spark平臺(tái)下的執(zhí)行時(shí)間對(duì)比 s

圖3 算法在Spark環(huán)境中的計(jì)算流程分析

2)Spark作業(yè)啟動(dòng)后會(huì)立即申請(qǐng)所需的Executor資源，并且所有Stage的Tasks以線程的方式運(yùn)行，共用Executors資源。這相對(duì)于MapReduce以心跳的方式管理slot資源，Spark申請(qǐng)資源的次數(shù)大大減少，導(dǎo)致資源管理效率高于MapReduce。

3)MapReduce中算法總共需要執(zhí)行多達(dá)12個(gè)Job，即使優(yōu)化后能夠減少到7個(gè)，但作業(yè)的數(shù)量仍然較多。而通過(guò)Spark中的DAG編程模型，可以實(shí)現(xiàn)將多個(gè)MapReduce作業(yè)簡(jiǎn)化為單個(gè)Spark DAG作業(yè)。而DAG作業(yè)會(huì)進(jìn)一步分解為多個(gè)Stage，每個(gè)Stage包含多個(gè)可并行執(zhí)行的Task，由于Spark資源管理效率比MapReduce高，使得Spark中的Task并行執(zhí)行效率比MapReduce更高。

5 結(jié)語(yǔ)

大數(shù)據(jù)的規(guī)模效應(yīng)給數(shù)據(jù)存儲(chǔ)、管理以及數(shù)據(jù)分析帶來(lái)了極大挑戰(zhàn)，高效率低成本的大數(shù)據(jù)處理技術(shù)成為學(xué)術(shù)界及工業(yè)界的研究熱點(diǎn)。隨著MapReduce生態(tài)系統(tǒng)的日趨完善，MapReduce逐漸成為工業(yè)與學(xué)術(shù)屆事實(shí)上的海量數(shù)據(jù)并行處理標(biāo)準(zhǔn)，但MapReduce的優(yōu)勢(shì)在于處理批處理作業(yè)。對(duì)于具有復(fù)雜業(yè)務(wù)處理邏輯的互聯(lián)網(wǎng)數(shù)據(jù)挖掘類作業(yè)，由于這些算法通常需要多個(gè)MapReduce作業(yè)協(xié)作完成，但多個(gè)作業(yè)之間嚴(yán)重的冗余磁盤讀寫及重復(fù)的資源申請(qǐng)操作，使得算法的性能較低。在此背景下，本文選擇ItemBased作為研究對(duì)象，首先對(duì)基于MapReduce的ItemBased協(xié)同過(guò)濾算法實(shí)現(xiàn)及性能缺陷進(jìn)行了分析；并在此基礎(chǔ)上提出利用Spark在迭代計(jì)算和內(nèi)存計(jì)算上的優(yōu)勢(shì)，提高算法的執(zhí)行效率的方法，進(jìn)而在Spark平臺(tái)下對(duì)ItemBased算法進(jìn)行了實(shí)現(xiàn)；最后通過(guò)對(duì)比實(shí)驗(yàn)，驗(yàn)證了Spark平臺(tái)下算法性能相對(duì)MapReduce平臺(tái)的優(yōu)越性，并對(duì)算法效率提高的原因進(jìn)行了分析。

下一步工作主要是在本文的基礎(chǔ)上，研究利用JVM(Java Virtual Machine)參數(shù)優(yōu)化、矩陣計(jì)算優(yōu)化及并行度調(diào)優(yōu)等方法進(jìn)一步提高算法的計(jì)算效率。

References)

[1] The digital universe in 2020: big data, bigger digital shadows, and biggest growth in the far east [EB/OL]. [2017- 03- 15]. http://www.emc.com/collateral/analyst-reports/idc-the-digitaluniverse-in - 2020.pdf.

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

[3] GHEMAWAT S, GOBIOFF H, LEUNG S T. The Google file system [C]// Proceedings of the 19th ACM Symposium on Operating System Principles. New York: ACM, 2003: 29-43.

[4] DEAN J, GHEMAWAT S. MapReduce: simplified data processing on large clusters [C]// OSDI 2004: Proceedings of the 2004 Conference on Operating System Design and Implementation. New York: ACM, 2004: 137-150.

[5] 廖彬,張?zhí)?于炯,等.MapReduce能耗建模及優(yōu)化分析[J].計(jì)算機(jī)研究與發(fā)展,2016,53(9):2107-2131.(LIAO B, ZHANG T, YU J, et al. Energy consumption modeling and optimization analysis for MapReduce [J]. Journal of Computer Research and Development, 2016, 53(9): 2107-2131.)

[6] 廖彬,于炯,張?zhí)?等.基于分布式文件系統(tǒng)HDFS的節(jié)能算法[J].計(jì)算機(jī)學(xué)報(bào),2013,36(5):1047-1064.(LIAO B, YU J, ZHANG T, et al. Energy-efficient algorithms for distributed file system HDFS [J]. Chinese Journal of Computers, 2013, 36(5): 1047-1064.)

[7] 張?zhí)?于炯,廖彬,等.基于GraphX的傳球網(wǎng)絡(luò)構(gòu)建及分析研究[J].計(jì)算機(jī)研究與發(fā)展,2016,53(12):2729-2752.(ZHANG T, YU J, LIAO B, et al. The construction and analysis of pass network graph based on GraphX [J]. Journal of Computer Research and Development, 2016, 53(12): 2729-2752.)

[8] 宋杰,劉雪冰,朱志良,等.一種能效優(yōu)化的MapReduce資源比模型[J].計(jì)算機(jī)學(xué)報(bào),2015,38(1):59-73.(SONG J, LIU X B, ZHU Z L, et al. An energy-efficiency optimized resource ratio model for MapReduce [J]. Chinese Journal of Computers, 2015, 38(1): 59-73.)

[9] 廖彬,張?zhí)?于炯,等.溫度感知的MapReduce節(jié)能任務(wù)調(diào)度策略[J].通信學(xué)報(bào),2016,37(1):61-75.(LIAO B, ZHANG T, YU J, et al. Temperature aware energy-efficient task scheduling strategies for MapReduce [J]. Journal on Communications, 2016, 37(1): 61-75.)

[10] 廖彬,張?zhí)?于炯,等.適應(yīng)節(jié)能與異構(gòu)環(huán)境的MapReduce數(shù)據(jù)布局策略[J].中山大學(xué)學(xué)報(bào)(自然科學(xué)版),2015,54(6):55-66.(LIAO B, ZHANG T, YU J, et al. An energy-efficient and heterogeneous environment adaptive data layout strategy for MapReduce [J]. Acta Scientiarum Naturalium Universitatis Sunyatseni, 2015, 54(6): 55-66.)

[11] 楊興耀,于炯,吐爾根·依布拉音,等.融合奇異性和擴(kuò)散過(guò)程的協(xié)同過(guò)濾模型.軟件學(xué)報(bào),2013,24(8):1868-1884.(YANG X Y, YU J, IBRAHIM T, et al. Collaborative filtering model fusing singularity and diffusion process [J]. Journal of Software, 2013, 24(8): 1868-1884.)

[12] GHAUTH K I, ABDULLAH N A. Learning materials recommendation using good learners’ ratings and content-based filtering [J]. Educational Technology Research and Development, 2010, 58(6): 711-727.

[13] UDDIN M N, SHRESTHA J, JO G S. Enhanced content-based filtering using diverse collaborative prediction for movie recommendation [C]// Proceedings of the 1st Asian Conference on Intelligent Information and Database Systems. Piscataway, NJ: IEEE, 2009: 132-137.

[14] NGUYEN A T, DENOS N, BERRUT C. Improving new user recommendations with rule-based induction on cold user data [C]// Proceedings of the 2007 ACM Conference on Recommender Systems. New York: ACM, 2007: 121-128.

[15] CHUN J, OH J Y, KWON S, et al. Simulating the effectiveness of using association rules for recommendation systems [C]// Proceedings of the 2005 Systems Modeling and Simulation: Theory and Applications. Berlin: Springer, 2005: 306-314.

[16] QIU L Y, BENBASAT I. A study of demographic embodiments of product recommendation agents in electronic commerce [J]. International Journal of Human-Computer Studies, 2010, 68(10): 669-688.

[17] CHEN T, HE L. Collaborative filtering based on demographic attribute vector [C]// Proceedings of the 2009 ETP International Conference on Future Computer and Communication. Piscataway, NJ: IEEE, 2009: 225-229.

[18] JIA C X, LIU R R, SUN D, et al. A new weighting method in network-based recommendation [J]. Physica A—Statistical Mechanics and Its Applications, 2008, 387(23): 5887-5891.

[19] ZHOUT, REN J, MEDO M, et al. Bipartite network projection and personal recommendation [J]. Physical Review E, 2007, 76(4): 1-7.

[20] LIU Z B, QU W Y, LI H T, et al. A hybrid collaborative filtering recommendation mechanism for P2P networks [J]. Future Generation Computer Systems, 2010, 26(8): 1409-1417.

[21] ZHAO Z D, SHANG M S. User-based collaborative-filtering recommendation algorithms on Hadoop [C]// Proceedings of the 2010 International Conference on Knowledge Discovery and Data Mining. Piscataway, NJ: IEEE, 2010: 478-481.

[22] MA M M, WANG S P. Research of user-based collaborative filtering recommendation algorithm based on Hadoop [C]// Proceedings of the 2015 International Conference on Computer Information Systems and Industrial Applications. Amsterdam: Atlantis Press, 2015: 63-66.

[23] SCHELTER S, BODEN C, MARKL V. Scalable similarity-based neighborhood methods with MapReduce [C]// Proceedings of the 2012 ACM Conference on Recommender Systems. New York: ACM, 2012: 163-170.

[24] DAS A S, DATAR M, GARG A, et al. Google news personalization: scalable online collaborative filtering [C]// Proceedings of the 2007 International Conference on World Wide Web. New York: ACM, 2007: 271-280.

[25] JIANG J, LU J, ZHANG G, et al. Scaling-up item-based collaborative filtering recommendation algorithm based on Hadoop [C]// Proceedings of the 2011 IEEE World Congress on Services. Piscataway, NJ: IEEE, 2011: 490-497.

This work is partially supported by the National Natural Science Foundation of China (61562078, 61262088), the Natural Science Foundation of Xinjiang Uygur Autonomous Region (2016D01B014).

LIAOBin, born in 1986, Ph. D., associate professor. His research interests include green computing, data mining, big data calculation model.

ZHANGTao, born in 1988, Ph. D. candidate. Her research interests include distributed computing, grid computing.

GUOBinglei, born in 1991, Ph. D. candidate. Her research interests include green computing, database system.

YUJiong, born in 1964, Ph. D., professor. His research interests include network security, grid computing, distributed computing.

ZHANGXuguang, born in 1994, M. S. candidate. His research interests include big data computing.

LIUYan, born in 1990, M. S. candidate. His research interests include big data computing.

PerformanceoptimizationofItemBasedrecommendationalgorithmbasedonSpark

LIAO Bin1*, ZHANG Tao2,3, GUO Binglei3, YU Jiong3, ZHANG Xuguang1, LIU Yan4

(1.CollegeofStatisticsandInformation,XinjiangUniversityofFinanceandEconomics,UrumqiXinjiang830012,China;2.CollegeofMedicalEngineeringandTechnology,XinjiangMedicalUniversity,UrumqiXinjiang830011,China;3.SchoolofInformationScienceandEngineering,XinjiangUniversity,UrumqiXinjiang830008,China;4.SchoolofSoftware,TsinghuaUniversity,Beijing100084,China)

Under MapReduce computing scenarios, complex data mining algorithms typically require multiple MapReduce jobs’ collaboration process to compete the task. However, serious redundant disk read and write and repeat resource request operations among multiple MapReduce jobs seriously degrade the performance of the algorithm under MapReduce. To improve the computational efficiency of ItemBased recommendation algorithm, firstly, the performance issues of the ItemBased collaborative filtering algorithm under MapReduce platform were analyzed. Secondly, the execution efficiency of the algorithm was improved by taking advantage of Spark’s performance superiority on iterative computation and memory computing, and the ItemBased collaborative filtering algorithm under Spark platform was implemented. The experimental results show that, when the size of the cluster nodes is 10 and 20, the running time of the algorithm in Spark is only 25.6% and 30.8% of that in MapReduce. The algorithm’s overall computing efficiency of Spark platform improves more than 3 times compared with that of MapReduce platform.

collaborative filtering; MapReduce; Spark algorithm; performance optimization; Directed Acyclic Graph (DAG)

TP393.09

:A

2017- 01- 16;

:2017- 03- 01。

國(guó)家自然科學(xué)基金資助項(xiàng)目(61562078, 61262088)；新疆維吾爾自治區(qū)自然科學(xué)基金資助項(xiàng)目(2016D01B014)。

廖彬(1986—)，男，四川內(nèi)江人，副教授，博士，CCF會(huì)員，主要研究方向：綠色計(jì)算、數(shù)據(jù)挖掘、大數(shù)據(jù)計(jì)算模型； 張?zhí)?1988—),女,新疆烏魯木齊人，博士研究生，主要研究方向：分布式計(jì)算、網(wǎng)格計(jì)算； 國(guó)冰磊(1991—)，女，湖北武漢人，博士研究生，主要研究方向:綠色計(jì)算、數(shù)據(jù)庫(kù)系統(tǒng)； 于炯(1964—)，男，北京人，教授，博士，主要研究方向:網(wǎng)絡(luò)安全、網(wǎng)格計(jì)算、分布式計(jì)算； 張旭光(1994—),男，河南鄭州人，碩士研究生，主要研究方向：大數(shù)據(jù)計(jì)算； 劉炎(1990—),男，湖北武漢人，碩士研究生，主要研究方向：大數(shù)據(jù)計(jì)算。

1001- 9081(2017)07- 1900- 06

10.11772/j.issn.1001- 9081.2017.07.1900