基于Spark優(yōu)化的協同過濾推薦算法研究

張嘉新，趙建平，蔣振剛

（長春理工大學 計算機科學技術學院，長春 130022）

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

1 ALS推薦算法

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

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

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

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

其中，rui表示第u個用戶對第i個物品的評分，xu表示用戶u的偏好隱含特征向量，yi表示商品i的隱含特征向量，xTuyi為用戶u對物品評分的近似，為了防止過擬合，加入正則化項：

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

同理固定X，由對稱性得：

反復進行以上兩步的計算，引入均方根誤差RMSE作為迭代終止的條件參數。

當均方根誤差RMSE的值變化很小，小于一個預設值時，就可以認為結果已經收斂，停止迭代，也可以預設迭代次數，當達到預設迭代次數時，迭代停止。

2 Spark優(yōu)化

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

如圖1所示，由于Spark集群是主從式分布結構，即在集群中只有一個Master節(jié)點負責整個集群中Worker節(jié)點的資源調度和分配，一旦Master節(jié)點出現異常就會導致整個集群無法工作，本文用Zookeeper技術來解決單點故障問題。

圖1 Spark主從結構圖

Zookeeper[12]是一個分布式的，開放源碼的分布式應用程序協調服務，它包含一個簡單的原語集，分布式應用程序可以基于它實現同步服務，配置維護和命名服務等。由Zookeeper維護的Spark集群與HDFS集群可以達到高可用的狀態(tài)（High Availability）。

圖2 Spark HA（High Availability）集群

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

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

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

圖3 無文件合并Shuffle流程

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

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

圖4 文件合并Shuffle流程

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

3 實驗及結果分析

3.1 實驗環(huán)境

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

表1 集群配置

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

表2 節(jié)點配置

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

表3 節(jié)點角色

3.2 實驗數據

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

3.3 實驗設計

解決針對新用戶的冷啟動問題：在數據集中選出最受歡迎的50部電影，在這50部電影中隨機選取10部電影，讓新用戶打分，根據用戶打分情況，生成用戶的特征向量，并使用ALS協同過濾推薦算法向用戶推薦50部電影。

通過三組實驗分別是無優(yōu)化，資源優(yōu)化與Shuffle優(yōu)化，觀察四項指標（RMSE：均方根誤差，Iteration：迭代次數，Rank：矩陣維度，Assessment：推薦基準線）與運行時間的關系。

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

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

4 結論

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

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