基于Spark框架的改進協同過濾算法

鄒紅旭，潘冠華，李 吟

(江蘇自動化研究所，江蘇 連云港 222006)

0 引 言

在互聯網產生的數據量迅速增長的今天，如何準確高效地挖掘數據中蘊藏的信息變得越來越有挑戰性[1]，推薦系統能有效地解決這一問題。推薦系統，是提供信息推薦的系統，以降低用戶尋找有效信息的難度[2]，增加用戶與網絡信息交互的體驗感為目的的算法本體。推薦系統始于二十世紀九十年代，歷經了二十多年的發展后，其理論也在不斷地更新和完善[3]，應用也變得越來越廣泛。然而，面對當今時代爆炸式增長的數據量，推薦算法的計算效率面臨很大的挑戰，單機已經無法滿足計算需求，因此，如何在多節點的計算機集群上高效地并行化運行推薦算法成為了重中之重[4]。

為了使協同過濾算法更能適應數據稀疏的情況，改進相似度計算公式，設計一種既能適應稀疏數據，又適合進行并行化改造的協同過濾算法，研究其在Spark框架下的并行化實現方案，并進行調優和驗證，使其準確度和計算效率達到最大化。文獻[5]研究了協同過濾算法在Spark上的并行化方案，但其在實現過程中采用了將數據集在集群上從單個節點進行廣播的方法，當數據集非常大時該方法會有明顯的瓶頸現象，進行廣播的節點會占用大量的計算時間，而且如果數據集大到無法在單一節點內存中存放時，該方法則無法完成計算。因此，提出了避免瓶頸問題的解決方案，采用了完全分布式的計算方法，每個節點始終只需存儲和計算一小部分數據，提高了系統的可擴展性。

1 Spark大數據計算框架

谷歌公司于2003年起依次發表了三篇論文：《Google文件系統(Google file system，GFS)》[6]、《大型結構化數據表(big table)》及《大數據分布式計算模型(MapReduce)》[7-8]。這三篇論文奠定了大數據技術的基礎，后來Apache基金會根據這三篇論文，設計了Hadoop分布式數據處理系統，從而將大數據的研究推向了高潮。作為MapReduce計算模型的替代方案，Apache Spark彌補了MapReduce在實現迭代式算法時由于多次進行磁盤I/O而降低算法計算效率、無法滿足算法實時性要求的缺陷[9]。

Spark采用了內存計算模型，對于迭代式算法，數據和計算中間結果都存儲在內存中，只需開始時從磁盤加載一次數據，之后都直接從內存中加載數據，省去了大量的磁盤I/O時間。Spark的系統架構如圖1所示，Spark運行程序時首先由Driver向集群管理器申請資源，包括CPU和內存資源，獲得集群管理器分配的資源之后，Driver將應用程序以task的形式發放到各個工作節點，工作節點并行執行各個task。

圖1 Spark的整體架構

2 改進的Item-Based協同過濾算法

2.1 Item-Based推薦算法原理

Item-Based推薦算法[10-13]的輸入為用戶-項目評分數據集，該數據集由數量為m的用戶對數量為n的項目的評分構成，用戶集合用U={u1,u2,…,um}表示，項目集合用I={i1,i2,…,in}表示。評分數據集是不完善的，每個用戶只評論了部分項目，推薦算法的目的就是完善該數據集，預測出所有用戶對所有項目的評分。計算過程如下：

(1)計算項目間相似度，得到相似度矩陣Simn×n。這里采用修正的余弦相似度計算方法，公式如下：

(1)

(2)根據相似度進行評分預測。使用Park采用的預測方法，計算公式如下：

2.2 算法改進思路

如式(1)所示，相似度的計算要用到兩個項目之間的公共用戶，即集合Uij里的用戶u分別對項目i、j的評分，但當數據稀疏時[14-16]，兩個項目之間的公共用戶的數目可能會很小，導致相似度的計算會出現很大的偏差。極端情況下，當Uij中只有一個元素時，根據式(1)計算得到的相似度為1，表示兩個項目間的相似度為100%，無論這兩個項目在實際中是否相似，都將得到這個結果。而在后續的計算中，這個值為1的相似度由于是最大值，又必然會被選入KNNi中，于是會對計算結果Pu,i的準確度產生很大的干擾。

針對上述問題，提出改進的相似度計算公式，在原公式(1)的基礎上乘以一個權重函數f(x)，x表示集合Uij中元素的數目，即兩項目間的公共用戶數目，目的是對當x很小時計算得出的相似度加一個很小的權重，使其盡量不被選入KNNi中，或者即使被選入KNNi中，也無法對Pu,i的計算產生很大的影響。另一方面，當兩個項目間公共用戶數目很少時，也能從一定程度上說明這兩個項目間的相似度很小，所以乘以權重函數f(x)之后得到的相似度將更加接近這兩個項目間的真實相似度。f(x)的選取要滿足以下兩個條件：

(1)f(x)是關于x的增函數；

(2)當x足夠大時f(x)隨x的增長要趨于收斂，因為此時根據式(1)計算得到的相似度已接近真實值，f(x)的值必須趨于穩定，才能保證加權之后對計算結果Pu,i產生的干擾達到最小。

考慮以上兩點，選取f(x)=ln(x)，改進后的相似度計算公式如下：

ln(x)

(3)

其中，x表示集合Uij中元素的數目。

3 改進的協同過濾算法在Spark上的并行化實現

3.1 計算資源的申請

采用的分布式環境為Spark+Yarn模式，Spark作為分布式計算框架，Yarn作為集群管理器，根據集群配置，Spark向Yarn申請10個executor作為計算資源，每個executor包括一個CPU與500 M內存。在Spark中，數據被切分為若干個分區，不同的分區位于不同的executor中，每個executor只需對自己分區中的數據進行計算，從而達到分布式計算的目的。分區數目的選擇是影響Spark計算效率的一個關鍵因素，分區數目太多會導致數據的混洗(shuffle)過程消耗更長的時間，而分區數目太少則會使join操作更加耗費時間并且降低系統容錯性。在本次實驗中分區數目選取為50個。計劃好資源的申請之后就可以用如下語句啟動spark：

spark-submit --master yarn --executor-memory 500 M --executor-cores 1 --num-executors 10

3.2 項目間相似度的計算

Spark主要通過把數據集抽象成RDD(resilient distributed dataset，彈性分布式數據集)對象來操作數據，通過對RDD的操作來間接操作數據。相似度的計算可以分為以下四個步驟：

(1)從HDFS(Hadoop distributed file system，Hadoop分布式文件系統)中讀取包含訓練數據的文件。需要用到數據文件中userId、itemId、rate這三個字段，分別表示用戶編號、項目編號、用戶對項目的評分。將數據利用textFile()方法讀入內存，并轉化為格式為(userId，(itemId，rate))的元組組成的RDDdata。

(2)依次利用map、reduceByKey、mapValues算子對RDDdata操作得到RDDavgrate，該RDD代表用戶評分均值，其元素的格式為(userId，averageRate)。因為后面要多次用到該RDD，為了避免Spark重復計算該RDD，利用persist操作將其持久化到內存中。

(3)用join算子對RDDdata和RDDavgrate進行等值連接，將得到的結果再次與RDDdata進行等值連接，得到RDDjoin，其元素的格式為(userId，(itemId1，rate1，itemId2，rate2，averageRate))，itemId1、itemId2表示該用戶共同評價過的兩個項目，rate1、rate2表示評分。

(4)對RDDjoin進行map操作，將其元素的格式轉換為(itemId1_itemId2,rate1,rate2，averageRate)，然后對其進行combineByKey操作得到RDDsimilarity，其元素的格式為(itemId1_itemId2,similarity)，similarity表示itemId1和itemId2之間的相似度。

相似度計算過程中用到的RDD的譜系圖如圖2所示。

圖2 相似度計算過程中的RDD譜系圖

3.3 評分預測

評分的預測可以分為以下幾個步驟：

(1)對RDDdata進行combineByKey操作，得到RDDitemList，其元素的格式為(userId,ArrayBuffer[(itemId,rate)])，其中ArrayBuffer表示scala語言中的變長數組，ArrayBuffer[(itemId,rate)]表示該用戶評價過的所有項目和對應評分組成的數組。

(2)從HDFS中讀取測試數據集RDDtest，并與RDDitemList進行join操作，再對結果進行map操作得到RDDtestList，其元素的格式為(itemId1,(userId,ArrayBuffer[(itemId2,rate)]))，其中itemId1、userId分別表示要進行預測評分的用戶和項目，需要注意的是itemId1來自測試集而itemId2來自訓練集。最后再對RDDtestList進行combineByKey聚合操作，使其鍵保持不變，值聚合到一個ArrayBuffer中，方便后續計算。

(3)依次利用map、reduceByKey、mapValues算子對RDDdata操作得到RDDitemavg，代表項目評分均值，其元素的格式為(itemId，itemAverageRate)。

(4)對RDDsimilarity和RDDitemavg進行join操作，再對結果進行flatMap、combineByKey操作得到RDDknn，代表項目的最近鄰居集，其元素的格式為(itemId1,ArrayBuffer[(itemId2,similarity,itemAverageRate)])。

(5)對RDDitemavg、RDDknn、RDDtestList進行join操作，再對結果進行mapValues、flatMap操作得到最終結果RDDresult，其元素格式為(itemId，userId，result)，result表示最終預測的評分。

評分預測過程中用到的RDD的譜系圖如圖3所示。

圖3 評分預測過程中的RDD譜系圖

3.4 等值連接操作的優化

根據上述步驟可知，計算過程中涉及到很多的join操作，而join操作會消耗很多的計算資源[17-18]，這主要是因為join操作會進行多次的迭代來匹配相同鍵值的元素，因此有必要對其進行優化。假設要對RDD1和RDD2進行連接操作，其元素數目分別為x1和x2，如果采用join算子會進行共x1*x2次迭代。設計自定義的Hash_Join函數替代join操作，Hash_Join函數原理如下：

假設RDD1的元素格式為(Key1,Value1)，RDD2的元素格式為(Key2,Value2)，首先對兩個RDD進行map操作，利用自定義的Hash函數對兩個RDD的Key進行轉換，使其映射到集合Buckets，原有的Key放入Value中。兩個RDD中相同的Key會被映射到相同的Bucket中。Key集合與Buckets集合是多對一的關系，目的是為了將多個Key對應的內容聚集到一個Bucket中。然后再對這兩個RDD進行join操作，把兩個RDD中相同Bucket的內容匹配到一起，最后再把每個Bucket中相同Key的內容匹配到一起。Hash_Join函數原理如圖4所示。

假設Key的數量是Bucket數量的a倍，則每個Bucket含有a條元素，優化后的迭代次數如下：

(4)

在對MovieLens數據集進行測試時，進行連接操作的數據集規模大小為百萬級，此時取a的大小為200，可以看出，優化后的迭代次數明顯小于x1*x2。

圖4 Hash_Join原理

4 實驗結果分析

實驗使用UCI的公用數據集MovieLens對算法進行測試，該數據集有三種不同大小的數據，其數據規模分別為100 k、1 M和10 M，實驗中采用規模為1 M的數據集，其中包含了6 040位用戶對3 900部電影的評分，評分記錄一共有1 000 209條。實驗用到的Spark集群部署在5臺服務器上，使用的集群管理器為Yarn管理器，通過向Yarn申請executor來獲取計算資源，實驗中申請的executor數目為10個，為每個executor分配一個CPU，CPU型號為Intel(R) Xeon(R) CPU E5-2630 v3 @2.40 GHz，每個executor分配的內存為500 M。

(1)算法準確度測試。

為了衡量預測評分的準確性，采用平均絕對誤差(MAE)來計算預測結果的誤差。先后把數據集分成不同比例的訓練集和測試集，分別用改進前的算法與改進后的算法對測試集進行預測，并分別對預測結果進行MAE的計算，結果如表1所示。

表1 不同的訓練集和測試集比例下兩種算法的MAE值

從結果中可以看出，改進后的算法預測結果的MAE要小于改進前的算法，說明改進后的算法提高了預測結果的準確度，并且可以看到，隨著訓練數據集的比例逐漸減小，即訓練數據越發稀疏時，兩種算法的MAE的差值逐漸增大，說明改進后的算法對預測準確度的提升愈加明顯，從而說明了改進后的算法更加適合數據稀疏的情況。

(2)算法執行時間測試。

在Spark集群上分別運行未優化的算法和優化等值連接操作后的算法，并與單機運行算法的時間進行對比，取訓練集與測試集比例為9∶1，數據集規模逐漸增加，實驗結果如表2所示。

表2 運行時間對比

從結果中可以看出，當數據規模較小時，集群算法的運行時間并沒有比單機算法短很多，加速比并不是很高，這主要是因為Spark啟動作業、分配任務等系統操作以及集群上各節點之間的通信、互相傳遞數據等操作占用了額外的運行時間，但隨著數據規模的擴大，加速比逐漸提高并趨于穩定。

經過等值連接優化的算法在數據規模較小時表現并不突出，因為在用自定義的Hash函數把兩個RDD的Key轉換為Bucket時要付出額外的計算成本，但隨著數據規模的擴大，其性能表現逐漸超過了未優化的算法，并且數據規模越大，其相比未優化的算法節省的計算時間越多。

5 結束語

設計了一種在Spark平臺上運行的改進的Item-Based協同過濾算法，使其更適合數據稀疏的情況，并對算法中涉及的等值連接操作進行了優化，提高算法效率。用MovieLens數據集對算法進行測試后得到的結果表明，算法在準確度和效率方面都有更好的表現，說明了算法既能更準確地得到預測結果，又能適應分布式計算平臺，更快地得到計算結果。