陸俊堯，李玲娟

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

0 引 言

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

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

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

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

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

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

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

(1)項(xiàng)目間相似度的計(jì)算方法。

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

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

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

該方法是通過(guò)將用戶對(duì)于項(xiàng)目的評(píng)分減去用戶評(píng)分均值來(lái)表現(xiàn)用戶對(duì)具體項(xiàng)目的評(píng)分與大眾評(píng)分的差異性。

cosadjusted(i,j)=

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

(2)評(píng)分計(jì)算方法。

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

(2)

基于最近鄰居集KNNi，利用評(píng)分公式得到目標(biāo)用戶對(duì)于項(xiàng)目i的評(píng)分，進(jìn)而完善用戶-項(xiàng)目評(píng)分矩陣，最后為用戶推薦評(píng)分較高的項(xiàng)目。

2 Spark大數(shù)據(jù)計(jì)算平臺(tái)

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

Step6：對(duì)目標(biāo)項(xiàng)目i進(jìn)行預(yù)測(cè)評(píng)分的計(jì)算。基于RDDsimilarity，利用filter操作，將RDDsimilarity中key值包含項(xiàng)目i的元組過(guò)濾出來(lái)，然后將其轉(zhuǎn)化為L(zhǎng)ist集合，對(duì)List集合按照相似度值進(jìn)行從大到小的排序，最后獲取排序后的List的前K個(gè)值，組成項(xiàng)目i的最近鄰居集合KNNi，KNNi也是一個(gè)List集合。

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

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

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

4 實(shí)驗(yàn)與結(jié)果分析

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

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

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

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

(2)評(píng)分預(yù)測(cè)準(zhǔn)確性測(cè)試。

衡量推薦系統(tǒng)推薦準(zhǔn)確度[14]的最重要指標(biāo)就是其評(píng)分預(yù)測(cè)準(zhǔn)確性，通常采用均方根誤差(RMSE)和平均絕對(duì)誤差(MAE)來(lái)計(jì)算，兩者可反映出真實(shí)值和預(yù)測(cè)值之差，兩者的值越小，說(shuō)明評(píng)分預(yù)測(cè)的準(zhǔn)確性就越高。

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

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

可以看出，運(yùn)行在Spark集群上的Item-Based協(xié)同過(guò)濾算法的MAE值較小，而且MAE值隨著訓(xùn)練集比例的提高而降低，即歷史數(shù)據(jù)越豐富則推薦準(zhǔn)確度也越高，這體現(xiàn)出了大數(shù)據(jù)計(jì)算框架的重要性，因?yàn)樗梢蕴幚砀嗟臍v史數(shù)據(jù)，從而帶來(lái)更高的精確度。

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

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

圖3 基于單機(jī)和Spark集群的運(yùn)行時(shí)間

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

5 結(jié)束語(yǔ)

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