陸俊堯，李玲娟

(南京郵電大學(xué) 計算機學(xué)院，江蘇 南京 210023)

0 引 言

推薦系統(tǒng)的目的是為用戶進行精準高效的信息推送，在現(xiàn)有的推薦技術(shù)中，基于協(xié)同過濾的推薦技術(shù)是最為成功的?！皡f(xié)同過濾”[1]一詞最早是由GlodBerg等在90年代中期開發(fā)推薦系統(tǒng)Tapestry[2]時提出的，并且在后來得到了廣泛的研究和應(yīng)用。協(xié)同過濾推薦算法主要分為基于用戶間相似度的(User-Based)[3-5]協(xié)同過濾推薦算法和基于項目間相似度的(Item-Based)[6-9]協(xié)同過濾推薦算法。前者基于用戶之間的相似度，為目標用戶推薦其相近用戶感興趣的項目；后者計算項目之間的相似度，從而為目標用戶提供與其感興趣項目相似度較高的項目。

由于與日益增加的用戶數(shù)量相比，項目的數(shù)量是相對穩(wěn)定的。所以，基于項目間相似度的協(xié)同過濾算法可以有效地減少計算量，提高推薦的時效性。但是面對爆炸式增長的數(shù)據(jù)量，協(xié)同過濾算法的計算效率仍面臨著挑戰(zhàn)。

作為新一代大數(shù)據(jù)計算平臺，Spark具有基于內(nèi)存計算的特性，同時也支持對加載到內(nèi)存中數(shù)據(jù)的反復(fù)查詢，非常適合大數(shù)據(jù)場景下的數(shù)據(jù)計算，已成為當(dāng)今最為火熱的大數(shù)據(jù)計算框架之一[10]。

為了提高協(xié)同過濾乃至推薦系統(tǒng)的效率，研究了Item-Based協(xié)同過濾算法在Spark平臺上的并行化方案，并通過實驗對該方案的準確性以及時效性進行了驗證。

1 Item-Based協(xié)同過濾算法原理分析

在基于協(xié)同過濾的推薦系統(tǒng)中，最關(guān)鍵的部分就是用戶-項目評分矩陣A(m,n)，它由數(shù)量為m的用戶集合U={u1,u2,…,um}對數(shù)量為n的項目集合I={i1,i2,…,in}的評分構(gòu)成，用戶u對項目i的評分用Rui表示，推薦則基于評分矩陣按照評分高低進行。而基于協(xié)同過濾的推薦算法的任務(wù)就是完善和精確這個評分矩陣。

Item-Based協(xié)同過濾算法完善用戶-項目評分矩陣的過程大致如下：首先計算項目間的相似度，得到相似度矩陣Simn×n；然后在完善用戶對未瀏覽項i的評分時，根據(jù)相似度矩陣Simn×n選取K個與i相似度最高的項目來組成最近鄰居集KNNi；再基于最近鄰居集KNNi利用評分公式進行計算，得到預(yù)測評分。

(1)項目間相似度的計算方法。

定義1:對?i∈I，定義項目-評分矩陣Am×n中對應(yīng)于i的列為項目i的評分向量，記為Ui。

定義2：對?u∈U，定義項目-評分矩陣Am×n中第u行對應(yīng)于用戶u的評分向量，記為Iu。

Item-Based協(xié)同過濾算法中項目間相似度的計算主要涉及到用戶對項目的評分向量，計算方法有標準的余弦相似度、修正的余弦相似度、相關(guān)相似度等。其中修正的余弦相似度計算方法如式1所示。

該方法是通過將用戶對于項目的評分減去用戶評分均值來表現(xiàn)用戶對具體項目的評分與大眾評分的差異性。

cosadjusted(i,j)=

基于相似度計算公式進行項目間相似度計算，生成項目相似度矩陣Simn×n。該矩陣由項目集合I={i1,i2,…,in}兩兩間的相似度Sim(i,j)組成。

(2)評分計算方法。

預(yù)測用戶對于未評分或者未瀏覽項目i的評分的方法包括加權(quán)相似度預(yù)測方法和Park采用的預(yù)測方法等，其中Park采用的預(yù)測方法的計算公式如下：

(2)

基于最近鄰居集KNNi，利用評分公式得到目標用戶對于項目i的評分，進而完善用戶-項目評分矩陣，最后為用戶推薦評分較高的項目。

2 Spark大數(shù)據(jù)計算平臺

Spark是新一代大數(shù)據(jù)計算平臺的代表，由UC Berkeley的AMPLab實驗室提出。相較于傳統(tǒng)大數(shù)據(jù)計算平臺Hadoop[11]，Spark基于內(nèi)存進行數(shù)據(jù)計算，原生語言采用了更為精煉簡潔的Scala，不僅可以進行數(shù)據(jù)的離線計算也可以進行實時計算，不僅可以利用自帶的資源調(diào)度框架運行(Standalone)，還可以利用Hadoop的資源調(diào)度框架Yarn、Mesos運行，具有速度快、易用、適用范圍廣、可擴展等特點。

Spark的核心機制包括彈性分布式數(shù)據(jù)集RDD和分布式運行架構(gòu)等。RDD是Spark中最為基本的數(shù)據(jù)抽象，代表一個由可分區(qū)、不可變、內(nèi)部元素可并行化計算的集合。首先，RDD由分區(qū)組成，分區(qū)是存取數(shù)據(jù)、進行計算的基本單位，其數(shù)量可由用戶按需求自定義。在進行計算時，Spark會根據(jù)分區(qū)中數(shù)據(jù)的物理位置進行計算的遷移，從而減少網(wǎng)絡(luò)中數(shù)據(jù)的傳輸，提升運算速度，分區(qū)的存在同時還提升了計算過程中的并發(fā)度。其次，在Spark中，程序計算過程的本質(zhì)就是不同RDD之間的轉(zhuǎn)換，計算過程中的中間值與最終結(jié)果均被保存于一系列的RDD之中。每個RDD都可以通過持久化操作被存儲于內(nèi)存或者磁盤上，而且RDD間類似于有向無環(huán)圖(DAG)的轉(zhuǎn)換依賴關(guān)系也會被保存起來。當(dāng)計算過程中發(fā)生錯誤導(dǎo)致RDD中數(shù)據(jù)丟失時，若該RDD進行過持久化操作，則可以直接進行調(diào)用恢復(fù)；否則，根據(jù)RDD中的依賴關(guān)系，從前往后重新進行計算，這在很大程度上提高了Spark的容錯性。

圖1 Spark的分布式運行架構(gòu)

Spark先進的分布式運行架構(gòu)如圖1所示。當(dāng)用戶向Spark集群提交一個計算任務(wù)時，驅(qū)動程序(Driver)會向資源管理器(Cluster Manager)申請資源。當(dāng)資源分配完成后，Spark便開始在工作節(jié)點(Worker)啟動多個任務(wù)執(zhí)行器(Executor)，然后等待Driver將規(guī)劃完成的計算任務(wù)集合(即RDD間轉(zhuǎn)換的步驟，類似于一個有向無環(huán)圖DAG)發(fā)送到Worker上。最終計算完成后，Worker將結(jié)果發(fā)回到Driver。

3 基于Spark的協(xié)同過濾算法的并行化方案設(shè)計

由于Item-Based協(xié)同過濾算法在計算過程中需要對數(shù)據(jù)進行反復(fù)的迭代計算，而Spark的RDD機制正好可以契合這個需求，因此，文中基于RDD轉(zhuǎn)換設(shè)計了基于Spark的Item-Based協(xié)同過濾算法的并行化方案，同時利用Spark中RDD緩存的特性來對一些計算消耗量極大的中間結(jié)果進行存儲，采用Spark的廣播變量的機制減少計算過程中的網(wǎng)絡(luò)數(shù)據(jù)傳輸[12]。方案的具體過程如下：

(1)Spark的配置與數(shù)據(jù)源的讀取。

Spark的驅(qū)動器程序會讀取相關(guān)的配置文件生成SparkConf對象，然后基于該對象進一步生成SparkContext對象去連接Spark集群。而Spark的計算流程就是通過讀取外部數(shù)據(jù)源或者Spark內(nèi)存中的數(shù)據(jù)將其轉(zhuǎn)換為源RDD，然后利用RDD的算子操作進行不同RDD間的轉(zhuǎn)換，最終得到結(jié)果RDD。在計算過程中，一個RDD分區(qū)就會生成一個計算任務(wù)。如果RDD分區(qū)數(shù)量與Spark集群為程序分配的計算資源不匹配，會導(dǎo)致Spark的并行化計算效率降低。因此，需要對源RDD的分區(qū)數(shù)量進行合理調(diào)整。

(2)多節(jié)點并行化執(zhí)行Item-Based協(xié)同的過濾算法。

基于上文的分析，Item-Based協(xié)同過濾算法的執(zhí)行過程被分為兩大部分：項目間相似度計算以及評分計算。具體過程如下所述，其中Step1～Step5為項目間相似度計算過程，Step6～Step7為評分計算過程。

Step1：Item-Based協(xié)同過濾算法需要涉及到用戶對項目評分的歷史記錄，需要其中的用戶編號userId、項目編號itemId、用戶對項目的評分rate三個字段。因此，先對原始數(shù)據(jù)進行預(yù)處理生成包含這三個字段的源文件。然后讀取源文件，進行字段切分，轉(zhuǎn)換成格式為(userId,itemId,rate)的元組組成的RDDsource。

Step2：采用修正的余弦相似度進行項目間相似度的計算，會涉及到每個用戶的評分均值。因此，獲取RDDsource中的useId和rate字段，用reduceByKey操作將具有相同userId的元組聚合到一起以后，用avgRateCalculate()這個自定義的平均分計算方法計算用戶評分均值并存入RDDavgRate。RDDavgRate由格式為(userId,avgRate)的元組組成。作為分布式計算框架，Spark默認會為多個并行操作中所用到的同一個變量分別進行發(fā)送，這會導(dǎo)致計算過程中網(wǎng)絡(luò)上存在大量的數(shù)據(jù)傳輸，影響計算效率。針對該問題，文中采用Spark的廣播變量機制來提高計算效率，將有關(guān)數(shù)據(jù)封裝成廣播變量的數(shù)據(jù)分發(fā)到各個計算節(jié)點進行保存且只發(fā)送一次，后續(xù)計算節(jié)點需要用到該數(shù)據(jù)時直接從本地獲取，不依賴網(wǎng)絡(luò)傳輸。由于RDDavgRate中的數(shù)據(jù)在后續(xù)計算過程中會被反復(fù)查詢，因此先用collect操作把RDDavgRate中的數(shù)據(jù)從各Worker節(jié)點匯總到Driver中，用toMap操作轉(zhuǎn)化為Map集合MapavgRate，再將其作為廣播變量廣播到各Worker節(jié)點。

Step3：獲取RDDsource中的userId、itemId、rate字段，將itemId與rate用“-”符號連接成“項目-評分”字段。再利用reduceByKey操作將具有相同userId的元組聚合到一起，構(gòu)成存儲用戶歷史評分信息的RDDhistRate。RDD由格式為(userId,itemId1-rate1，itemId2-rate2，…，itemIdN-rateN)的元組構(gòu)成，N為編號為userId的用戶評論過的項目數(shù)量。由于后續(xù)的評分計算過程中同樣還涉及到對用戶歷史評分的查詢，因此利用toMap操作轉(zhuǎn)化為MaphistRate后，再將其封裝為廣播變量，分發(fā)到各個Worker節(jié)點。

Step4：基于RDDhistRate，對于每個userId，將其對應(yīng)的由“itemId-rate”字段組成的集合進行兩兩間類似笛卡爾積的合并。先將格式為(userId,itemId1-rate1，itemId2-rate2，…，itemIdN-rateN)的元組利用map操作變成由格式為(itemIdi-itemIdj,ratei-ratej)，i,j∈[1,N]的元組組成的List集合。接著利用flatMap操作將List集合中元素取出，再用reduceByKey操作進行聚合，最終得到被用戶同時評論過的兩個項目的評分匯總，存放于RDDrateSum之中。RDDrateSum由格式為(itemIdiitemIdj，ratei-ratej)，i,j∈[1,N]的元組組成。

Step5：以RDDrateSum和MapavgRate為參數(shù)，利用自定義的函數(shù)similarityCalculate()進行評分計算，最終得到包含所有項目之間相似度的結(jié)果RDDsimilarity。該RDD由格式為(itemIdiitemIdj,Sim(i,j))，i,j∈[1,N]的元組組成，其中M為項目的總數(shù)量。由于相似度的計算消耗大量資源，故采取Spark中RDD的緩存機制將RDDsimilarity緩存起來，避免后續(xù)計算中因數(shù)據(jù)丟失而產(chǎn)生的重復(fù)計算。由于后續(xù)計算過程中也要對RDDsimilarity中的數(shù)據(jù)反復(fù)查詢，因此也將RDDsimilarity封裝成廣播變量發(fā)送到每一個Worker節(jié)點。

Step6：對目標項目i進行預(yù)測評分的計算?；赗DDsimilarity，利用filter操作，將RDDsimilarity中key值包含項目i的元組過濾出來，然后將其轉(zhuǎn)化為List集合，對List集合按照相似度值進行從大到小的排序，最后獲取排序后的List的前K個值，組成項目i的最近鄰居集合KNNi，KNNi也是一個List集合。

Step7：以KNNi以及MaphistRate為輸入?yún)?shù)，利用自定義函數(shù)rateCalculate()進行預(yù)測評分的計算。該函數(shù)使用Park采用的預(yù)測方法作為計算公式進行評分計算，最終獲得用戶u對未評分的項目i的預(yù)測評分Rui。

以上過程中涉及到的RDD轉(zhuǎn)換流程如圖2所示。

圖2 RDD的轉(zhuǎn)換流程

4 實驗與結(jié)果分析

為了測試和驗證基于Spark的Item-Based協(xié)同過濾算法的準確性與時間效率，采用MovieLens數(shù)據(jù)集對基于Spark的并行Item-Based協(xié)同過濾算法進行性能測試。首先利用不同比例的測試集和訓(xùn)練集對該算法的準確性進行測試。然后，分別在單機和Spark集群上運行相同規(guī)模的數(shù)據(jù)集，進行算法時效性的對比實驗。用Scala語言進行算法實現(xiàn)。

(1)實驗環(huán)境和數(shù)據(jù)。

實驗配置的Spark環(huán)境包含了三個節(jié)點，一臺驅(qū)動節(jié)點Master，兩臺執(zhí)行節(jié)點Worker。每個節(jié)點的CPU版本信息為Intel CORE i5-4210H，每個CPU都擁有兩個處理單元，硬盤的讀寫速度為600.00 MB/s，Master節(jié)點配有6 G內(nèi)存，其余Worker節(jié)點為4 G。Spark版本為1.6.1；Spark運行的操作系統(tǒng)為CentOS 6.5；Java版本為JDK1.7.0_13；Scala版本為2.10.4。

實驗使用了GroupLens Research提供的MovieLens數(shù)據(jù)集[13]，該數(shù)據(jù)集提供了三種大小不同的數(shù)據(jù)集，分別為100 k，1 M和10 M。實驗采用了大小為100 k的數(shù)據(jù)集，其中包含943位用戶對1 682部電影共計10萬條評分記錄。記錄中包含user id、item id、rating、timestamp四個字段，分別表示用戶編號、電影編號、電影評分、評分時間戳。

(2)評分預(yù)測準確性測試。

衡量推薦系統(tǒng)推薦準確度[14]的最重要指標就是其評分預(yù)測準確性，通常采用均方根誤差(RMSE)和平均絕對誤差(MAE)來計算，兩者可反映出真實值和預(yù)測值之差，兩者的值越小，說明評分預(yù)測的準確性就越高。

實驗采用不同比例的訓(xùn)練集和測試集，利用MAE值進行準確度計算。結(jié)果如表1所示。

表1 不同的訓(xùn)練集和測試集比例下的MAE值

可以看出，運行在Spark集群上的Item-Based協(xié)同過濾算法的MAE值較小，而且MAE值隨著訓(xùn)練集比例的提高而降低，即歷史數(shù)據(jù)越豐富則推薦準確度也越高，這體現(xiàn)出了大數(shù)據(jù)計算框架的重要性，因為它可以處理更多的歷史數(shù)據(jù)，從而帶來更高的精確度。

(3)算法執(zhí)行時間測試。

分別在單機和Spark集群上針對算法運行時間做了測試實驗。訓(xùn)練集和測試集比例為9∶1，數(shù)據(jù)集樣本數(shù)量逐漸遞增。實驗結(jié)果如圖3所示。

圖3 基于單機和Spark集群的運行時間

由實驗結(jié)果可知，在數(shù)據(jù)量較小時，由于Spark的啟動需要消耗大量資源以及時間，因而無法體現(xiàn)并行化算法在時間效率方面的優(yōu)勢，但隨著數(shù)據(jù)量的增加，其時間效率明顯提升。

5 結(jié)束語

設(shè)計了一種Item-Based協(xié)同過濾算法在Spark集群中的并行化方案，并通過基于MovieLens數(shù)據(jù)集的實驗結(jié)果證明，在應(yīng)對大規(guī)模數(shù)據(jù)處理時，基于Spark的并行化Item-Based協(xié)同過濾算法，不僅可以保證評分的準確性，而且算法執(zhí)行速度更快，可以提高推薦系統(tǒng)的時效性。