張嘉新，趙建平，蔣振剛

（長(zhǎng)春理工大學(xué) 計(jì)算機(jī)科學(xué)技術(shù)學(xué)院，長(zhǎng)春 130022）

隨著信息技術(shù)和互聯(lián)網(wǎng)的發(fā)展，人們逐漸從信息匱乏的時(shí)代走入了信息過(guò)載的時(shí)代。在這個(gè)時(shí)代，無(wú)論是信息消費(fèi)者還是信息生產(chǎn)者都遇到了很大的挑戰(zhàn)，作為信息消費(fèi)者，如何從大量信息中找到自己感興趣的信息是一件非常困難的事情，作為信息生產(chǎn)者，如何讓自己生產(chǎn)的信息脫穎而出，受到廣大用戶的關(guān)注，也是一件非常困難的事情。推薦系統(tǒng)就是解決這一矛盾的重要工具。推薦系統(tǒng)的任務(wù)就是聯(lián)系用戶和信息，一方面幫助用戶發(fā)現(xiàn)對(duì)自己有價(jià)值的信息，另一方面讓信息能夠展現(xiàn)在對(duì)它感興趣的用戶面前，從而實(shí)現(xiàn)信息消費(fèi)者和信息產(chǎn)生者的雙贏。推薦算法能夠?qū)⒖赡苁芟埠玫馁Y訊或?qū)嵨铮ɡ纾弘娪?、電視?jié)目、音樂(lè)、書(shū)籍、新聞、圖片、網(wǎng)頁(yè)）推薦給使用者。推薦算法中，協(xié)同過(guò)濾算法是目前使用最多的一種算法?！皡f(xié)同過(guò)濾”最初的應(yīng)用是于 1992 年由Goldberg、Nicols、Oki及Terry提出[1]，應(yīng)用于Xerox公司在Paloma Alto研究中心資訊過(guò)載的問(wèn)題，最初的推薦系統(tǒng)Tapestry只是處理公司內(nèi)部郵件，數(shù)據(jù)量較少。隨著互聯(lián)網(wǎng)的飛速發(fā)展，一些電商一天的數(shù)據(jù)量超過(guò)50TB，其中有4億條產(chǎn)品信息和2億多名用戶的活動(dòng)信息，這無(wú)疑給推薦算法與大數(shù)據(jù)處理帶來(lái)了挑戰(zhàn)，基于模型ALS（Alternating Least Squares）協(xié)同過(guò)濾算法提供了良好的降維方法，而最新一代的大數(shù)據(jù)處理引擎Spark非常適合處理ALS的復(fù)雜迭代計(jì)算，以逐漸取代Hadoop大數(shù)據(jù)計(jì)算平臺(tái)[2-6]。Spark雖然計(jì)算速度快，但在數(shù)據(jù)量過(guò)大時(shí)，其效率很難達(dá)到滿意程度，本文基于Spark底層運(yùn)行機(jī)制與ALS算法特點(diǎn)對(duì)Spark作業(yè)進(jìn)行資源優(yōu)化與Shuffle優(yōu)化，通過(guò)合理分配資源與合并Shuffle過(guò)程產(chǎn)生的文件，達(dá)到降低內(nèi)存存儲(chǔ)壓力與減少磁盤(pán)I/O的目的，從減少了推薦時(shí)間提升了推薦效率。

1 ALS推薦算法

基于隱語(yǔ)義模型[7]的推薦算法所研究的核心問(wèn)題就是如何將數(shù)據(jù)抽象出的高維矩陣進(jìn)行矩陣分解的問(wèn)題，它本質(zhì)上是對(duì)用戶與商品的關(guān)系矩陣進(jìn)行降維。本文針對(duì)基于ALS模型的推薦算法進(jìn)行優(yōu)化研究。

ALS，譯為交替最小二乘法。將User-Item-Rating（用戶-商品-評(píng)分）的數(shù)據(jù)集建立一個(gè)User×Item（用戶-商品）的m×n矩陣，由于不是每個(gè)用戶都對(duì)每個(gè)商品進(jìn)行過(guò)評(píng)分，所以這個(gè)m×n的矩陣往往是稀疏的，因此ALS的核心就是將用戶對(duì)商品的評(píng)分矩陣分解為2個(gè)低秩矩陣，分別是用戶對(duì)商品隱含特征的偏好矩陣，和商品所包含的隱含特征矩陣，在矩陣分解的過(guò)程中，缺失的評(píng)分項(xiàng)得到了補(bǔ)充，基于這些補(bǔ)充的評(píng)分就可以給用戶進(jìn)行商品推薦了。

對(duì)于矩陣R(m×n)，ALS旨在找到兩個(gè)低秩矩陣X(m×k)和矩陣Y(k×n)，來(lái)近似的逼近R(m*n)，即

其中，R(m×n)代表用戶對(duì)商品的品分矩陣，X(m×k)代表用戶對(duì)隱含特征的偏好矩陣，Y(k×n)表示商品所包含隱含特征的矩陣，其中k＜＜min（m，n），為了使得矩陣X和Y的乘積盡可能地逼近R，采用最小化平方誤差損失函數(shù)：

其中，rui表示第u個(gè)用戶對(duì)第i個(gè)物品的評(píng)分，xu表示用戶u的偏好隱含特征向量，yi表示商品i的隱含特征向量，xTuyi為用戶u對(duì)物品評(píng)分的近似，為了防止過(guò)擬合，加入正則化項(xiàng)：

xu和yi耦合在一起，并不好求解，故引用ALS，先固定Y，將損失函數(shù)L(X,Y)對(duì)xu求偏導(dǎo)，即：

同理固定X，由對(duì)稱性得：

反復(fù)進(jìn)行以上兩步的計(jì)算，引入均方根誤差RMSE作為迭代終止的條件參數(shù)。

當(dāng)均方根誤差RMSE的值變化很小，小于一個(gè)預(yù)設(shè)值時(shí)，就可以認(rèn)為結(jié)果已經(jīng)收斂，停止迭代，也可以預(yù)設(shè)迭代次數(shù)，當(dāng)達(dá)到預(yù)設(shè)迭代次數(shù)時(shí)，迭代停止。

2 Spark優(yōu)化

2.1 單點(diǎn)故障優(yōu)化

如圖1所示，由于Spark集群是主從式分布結(jié)構(gòu)，即在集群中只有一個(gè)Master節(jié)點(diǎn)負(fù)責(zé)整個(gè)集群中Worker節(jié)點(diǎn)的資源調(diào)度和分配，一旦Master節(jié)點(diǎn)出現(xiàn)異常就會(huì)導(dǎo)致整個(gè)集群無(wú)法工作，本文用Zookeeper技術(shù)來(lái)解決單點(diǎn)故障問(wèn)題。

圖1 Spark主從結(jié)構(gòu)圖

Zookeeper[12]是一個(gè)分布式的，開(kāi)放源碼的分布式應(yīng)用程序協(xié)調(diào)服務(wù)，它包含一個(gè)簡(jiǎn)單的原語(yǔ)集，分布式應(yīng)用程序可以基于它實(shí)現(xiàn)同步服務(wù)，配置維護(hù)和命名服務(wù)等。由Zookeeper維護(hù)的Spark集群與HDFS集群可以達(dá)到高可用的狀態(tài)（High Availability）。

圖2 Spark HA（High Availability）集群

圖2所示，在HA集群中，可設(shè)置多個(gè)Standby狀態(tài)的Master節(jié)點(diǎn)，當(dāng)Master節(jié)點(diǎn)發(fā)生異常時(shí)，Standby Master會(huì)切換為Master狀態(tài)，處于Standby狀態(tài)的Master在接收到org.apache.spark.deploy.master.ZooKeeperLeaderElectionAgent發(fā)送的ElectedLeader消息后，就開(kāi)始通過(guò)ZK中保存的Application，Driver和Worker的元數(shù)據(jù)信息進(jìn)行故障恢復(fù)。

2.2 Spark Shuffle分析與優(yōu)化

由于ALS算法多次迭代的特點(diǎn)，Spark作業(yè)的性能主要消耗在了Shuffle階段，同時(shí)該環(huán)節(jié)還包含了大量的磁盤(pán)I/O、序列化、網(wǎng)絡(luò)傳輸?shù)炔僮?，因此要讓ALS推薦算法的性能提高，就要對(duì)Shuffle過(guò)程進(jìn)行優(yōu)化。Shuffle，譯為“洗牌”，在MapReduce過(guò)程中需要各節(jié)點(diǎn)的同一類數(shù)據(jù)匯集到某一節(jié)點(diǎn)進(jìn)行計(jì)算[11]，把這些分布在不同節(jié)點(diǎn)的數(shù)據(jù)按照一定的規(guī)則聚集到一起的過(guò)程稱為Shuffle。

圖3 無(wú)文件合并Shuffle流程

如圖3所示，該圖為Spark集群中的某個(gè)節(jié)點(diǎn)的架構(gòu)圖，此節(jié)點(diǎn)上運(yùn)行了4個(gè)ShuffleMapTask，節(jié)點(diǎn)的CPU Core個(gè)數(shù)為2。每個(gè)ShuffleMapTask都會(huì)為每個(gè)ResultTask創(chuàng)建一份BK（Bucket）緩存，以及對(duì)應(yīng)的SBF（ShuffleBlockFile）文件磁盤(pán)文件（下文統(tǒng)稱Bucket與ShuffleBlockFile為輸出文件），ShuffleMapTask的輸出會(huì)作為MapStatus發(fā)送到DAGScheduler的MapOutTrackerMaster中，MapStatus包含了每個(gè)ResultTask要拉取數(shù)據(jù)的大小，每個(gè)ResultTask會(huì)用BlockStoreShuffleFetcher去MapOutputTrackerMaster獲取自己要拉取的數(shù)據(jù)信息，然后通過(guò)BlockManager將數(shù)據(jù)拉取過(guò)來(lái)。每個(gè)Result-Task拉取過(guò)來(lái)的數(shù)據(jù)會(huì)組成一個(gè)內(nèi)部RDD，即ShuffleRDD，如果內(nèi)存不夠，則寫(xiě)入磁盤(pán)，隨后每個(gè)ResultTask針對(duì)數(shù)據(jù)進(jìn)行聚合，生成MapPartitionsRDD。假設(shè)有100個(gè)MapTask，100個(gè)ResultTask，那么本地磁盤(pán)要產(chǎn)生10000個(gè)文件，由此可以看出Spark的Shuffle過(guò)程會(huì)消耗大量的磁盤(pán)I/O（輸入輸出）資源，嚴(yán)重影響推薦性能。

通過(guò)優(yōu)化ConsolidateFiles，即map端輸出文件合并機(jī)制，可以解決map端輸出文件過(guò)多的問(wèn)題，如圖4所示。

圖4 文件合并Shuffle流程

在并行運(yùn)行了ShuffleMapTask、后，每個(gè)ShuffleMapTask都創(chuàng)建了4個(gè)BK緩存，以及對(duì)應(yīng)的SBF磁盤(pán)文件，ShuffleMapTask、執(zhí)行完成后，Shuffle-MapTask、在并行運(yùn)行時(shí)不會(huì)重新創(chuàng)建新的輸出文件，而是復(fù)用之前的ShuffleMapTask、創(chuàng)建的輸出文件，并將數(shù)據(jù)寫(xiě)入ShuffleMapTask、的輸出文件中，ResultTask在拉取數(shù)據(jù)時(shí)，只是拉取少量數(shù)據(jù)，每個(gè)輸出文件中可能包含了多個(gè)ShuffleMapTask給自己的輸出文件，這樣輸出文件數(shù)量就減少了一倍。在實(shí)際應(yīng)用中假設(shè)有100個(gè)節(jié)點(diǎn)，每個(gè)節(jié)點(diǎn)一個(gè)Executor，每個(gè)Executor分配 2個(gè)CPU Core，分別有1000個(gè)ShuffleMapTask和ResultTask，將會(huì)產(chǎn)生100萬(wàn)個(gè)輸出文件，磁盤(pán)對(duì)I/O的壓力非常大。在優(yōu)化ConsolidateFiles后，每個(gè)Executor執(zhí)行10個(gè)ShuffleMapTask，那么每個(gè)節(jié)點(diǎn)輸出文件的數(shù)量是2000個(gè)，100個(gè)節(jié)點(diǎn)輸出文件的總數(shù)減少至20萬(wàn)個(gè)，磁盤(pán)I/O壓力減小了5倍。

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

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

本文中所搭建的Spark大數(shù)據(jù)集群為高可用集群，即在集群服務(wù)器架構(gòu)中，當(dāng)主服務(wù)器故障時(shí)，備份服務(wù)器能夠自動(dòng)接管主服務(wù)器的工作，并及時(shí)切換過(guò)去，以實(shí)現(xiàn)對(duì)用戶的不間斷服務(wù)，本文所使用的存儲(chǔ)系統(tǒng)是Hadoop中的HDFS（分布式文件系統(tǒng)），具體版本如表1所示。

表1 集群配置

Spark大數(shù)據(jù)集群的優(yōu)勢(shì)就是可以不用專業(yè)的、價(jià)格高昂的服務(wù)器，而用普通的個(gè)人電腦即可，本文實(shí)驗(yàn)中所用的計(jì)算機(jī)配置如表2所示。

表2 節(jié)點(diǎn)配置

如表3所示，節(jié)點(diǎn)glance02、glance03為Master節(jié)點(diǎn)，即在程序運(yùn)行中，一旦其中一個(gè)Active狀態(tài)的Master節(jié)點(diǎn)出現(xiàn)故障，另一個(gè)Standby狀態(tài)的Master節(jié)點(diǎn)會(huì)以秒級(jí)的速度切換為Active狀態(tài)，使程序繼續(xù)運(yùn)行。

表3 節(jié)點(diǎn)角色

3.2 實(shí)驗(yàn)數(shù)據(jù)

本文所使用的數(shù)據(jù)集為MovieLens數(shù)據(jù)集，它是由明尼蘇達(dá)大學(xué)的GroupLens研究組組織收集的，數(shù)據(jù)集中包括用戶數(shù)據(jù)，電影數(shù)據(jù)與電影的評(píng)分?jǐn)?shù)據(jù)，主要用于推薦系統(tǒng)的研究，數(shù)據(jù)集包含幾種級(jí)別數(shù)量級(jí)的數(shù)據(jù)，本文使用的數(shù)據(jù)大小為：電影數(shù)據(jù)10618條，電影評(píng)分?jǐn)?shù)據(jù)為1000萬(wàn)條。

3.3 實(shí)驗(yàn)設(shè)計(jì)

解決針對(duì)新用戶的冷啟動(dòng)問(wèn)題：在數(shù)據(jù)集中選出最受歡迎的50部電影，在這50部電影中隨機(jī)選取10部電影，讓新用戶打分，根據(jù)用戶打分情況，生成用戶的特征向量，并使用ALS協(xié)同過(guò)濾推薦算法向用戶推薦50部電影。

通過(guò)三組實(shí)驗(yàn)分別是無(wú)優(yōu)化，資源優(yōu)化與Shuffle優(yōu)化，觀察四項(xiàng)指標(biāo)（RMSE：均方根誤差，Iteration：迭代次數(shù)，Rank：矩陣維度，Assessment：推薦基準(zhǔn)線）與運(yùn)行時(shí)間的關(guān)系。

圖5 不同優(yōu)化方式的時(shí)間-參數(shù)對(duì)比圖

其中，圖5（a）是三組實(shí)驗(yàn)中RMSE的變化值與運(yùn)行時(shí)間的關(guān)系；圖5（b）是三組實(shí)驗(yàn)的最佳迭代次數(shù)與運(yùn)行時(shí)間的關(guān)系；圖5（c）是三組實(shí)驗(yàn)的最佳矩陣維度與運(yùn)行時(shí)間的關(guān)系，圖5（d）中Assessment是高出推薦基準(zhǔn)評(píng)估線的百分比，由這三組實(shí)驗(yàn)可以看出，在RMSE與Assessment的值基本不變的情況下，通過(guò)資源配置的改進(jìn)推薦時(shí)間由21分鐘縮短至14分鐘，提升效率為33.3%，通過(guò)Shuffle的map端文件合并的優(yōu)化，推薦時(shí)間由14分鐘縮短至9.5分鐘提升的總效率為54.8%。

4 結(jié)論

針對(duì)于應(yīng)用行性強(qiáng)的推薦算法，性能的提高不能僅僅需要考慮推薦算法本身，在數(shù)據(jù)量急劇增加今天，大數(shù)據(jù)處理技術(shù)的提高與優(yōu)化也是不可或缺的，本文提出的Shuffle過(guò)程中map端的文件輸出合并優(yōu)化是實(shí)際生產(chǎn)中容易忽視而又重要的問(wèn)題，通過(guò)實(shí)驗(yàn)證明，本文提供的改進(jìn)算法是可行、有效的。

Spark是基于內(nèi)存計(jì)算的快速的大數(shù)據(jù)計(jì)算引擎，要想使推薦效率進(jìn)一步提高，在Spark的優(yōu)化上還有很多值得研究的地方，接下來(lái)可以從分布式數(shù)據(jù)的網(wǎng)絡(luò)傳輸，ShuffleMapTask的執(zhí)行并行度及實(shí)際項(xiàng)目重構(gòu)RDD架構(gòu)等方面進(jìn)行更深一步的研究。