基于大數據集的協同過濾算法的并行化研究

李 改，潘 嶸，李章鳳，李 磊

（1.順德職業技術學院 電子與信息工程系，廣東 順德528300；2.中山大學 信息科學與技術學院，廣東 廣州510006；3.中山大學 軟件研究所，廣東 廣州510275）

0 引 言

協同過濾算法是推薦系統中運用最廣泛的的推薦算法［1－3］。協同過濾算法的核心是分析用戶興趣，在用戶群中找到與指定用戶相似（興趣）的用戶，綜合這些相似用戶對某一信息的評價，形成系統對該指定用戶對此信息的喜好程度預測［4－5］。最近幾年提出了各種高效的CF算法，其中包括潘嶸等提出了基于ALS的協同過濾推薦算法［6－8］，N.Srebro等提出了 MMMF［9－10］，R.Salakhutdinov等提出了PMF和 RBM［11－12］，Daniel D.Lee等提出了 NNMF［13］，以及聚類模型等等。

當前對這些協同過濾算法的研究都側重于單節點的算法設計與實現。但隨著互聯網的迅猛發展，推薦系統中用戶及推薦對象的數量在呈幾何倍數增長，使得實現在單節點機器上的這些算法要算出結果需要耗費大量時間，無法滿足大數據集的運算需要。因此如果我們能對這些算法實現分布式計算，將會大大縮短計算所需時間，同時必將對大規模協同過濾算法的應用研究有較大的推動作用。

本文的主要貢獻是：研究基于矩陣分解的Alternating－Least－Squares（ALS）協同過濾算法的并行化問題，并詳細介紹如何在開源的云計算平臺Hadoop［14］上實現該算法的并行化。同時對ALS算法在多個節點下的并行化算法與其在單節點上的串行算法運行的時間進行對比，進而對實驗進行評估。實驗證明了ALS算法的可并行性，并行后ALS算法的運算性能獲得了極大提高。

1 基于ALS協同過濾算法介紹

本節我們將首先簡單介紹基于ALS的二維協同過濾推薦算法。

給定一個矩陣（R＝（Rij）m×n）∈ ｛0，1｝m×n，該矩陣表示具有m個用戶、n個對象的評分矩陣。我們希望找到一個低秩矩陣X，來逼近矩陣R。同時最小化下面的Frobenius損失函數

在上面的目標函數L（X）中，（Rij－Xij）2是低秩逼近中常見平方誤差項。下面我們來考慮如何有效并且快速的求解最優化問題argminXL（X）。

考慮矩陣分解X＝UVT，U∈Cm×d，d表示特征個數，一般d＜＜r，r表示矩陣R的秩，r≈min（m，n）。這時式（1）可以改寫為

為了防止過擬合，我們給式（2）加上正則化項，則式（2）可改寫為

固定V，對Ui求導我們得到下面求解 Ui.的公式

式（4）中的Ri.表示用戶i評過的電影的評分組成的向量，Vui表示用戶i評過的電影的特征向量組成的特征矩陣。nui表示用戶i評過的電影的數量。

同理，固定U，可以得到下面求解Vj.的公式

式（5）中的R.j表示評過電影j的用戶的評分組成的向量，Umj表示評過電影j的用戶的特征向量組成的特征矩陣。nmj表示評過電影j的用戶的數量。

在式（4）、（5）中I表示一個d×d的單位矩陣。

基于式（4）、（5），我們提出下面的基于代正則化的交叉最小二乘法（ALS）的二維協同過濾推薦算法。①我們用0均值，偏差為0.01的高斯隨機數初始化矩陣V，②我們用式（4）更新矩陣U，接著我們用式（5）更新矩陣V，直到本算法計算出的RMSE值收斂或迭代次數足夠多而結束迭代為止。具體算法描述如下：

算法1 基于ALS的二維協同過濾推薦算法

輸入：用戶的評分矩陣R，特征個數d。

輸出：矩陣R的逼近矩陣X。

（1）初始化V。

（2）反復迭代運用式（4）、（5）更新U、V，直到本算法計算出的RMSE值收斂或迭代次數足夠多而結束迭代。

（3）X＝UVT，返回矩陣X。

2 ALS算法在Hadoop上并行實現

本節主要介紹ALS算法在Hadoop上的并行實現。包括算法設計和算法的實現細節部分。

2.1 Hadoop平臺簡介

云計算的核心計算模式是Map/Reduce，該技術是云計算的運算基礎。它的存儲基礎是 Hadoop Distributed File System（HDFS）項目，是云平臺的大規模數據存儲技術。下面我們就這兩個技術分別加以介紹［14－15］。

Hadoop的計算核心是Map/Reduce模式［16］。區別于網格計算，Map/Reduce計算模式要求并行處理的數據塊之間是相互獨立的。Map/Reduce計算模式的數據輸入是Key/Value對。Map過程由Key/Value進行數據的處理和整合，輸出到Reduce過程，最終的輸出依然是Key/Value數據對，Map和Reduce操作由程序員自己編寫提交給系統處理。由Hadoop的作業調度機制將數據和Map和Reduce操作分發給不同的虛擬機進行處理，并把計算結果輸出到分布式文件存儲平臺上。計算集群中的各個數據處理模塊相互獨立。Map過程處理完成之后，系統要對結果進行排序，排序后的結果又由Hadoop的作業調度機制分發給不同Reduce操作處理，最終將結果輸出。MapReduce的處理流程圖可參考文獻 ［14－15］。

HDFS是一個高度容錯的文件系統，非常適合部署到廉價的機器群上。HDFS的設計側重于數據的吞度量上，而不是處理速度。尤其是與Hadoop的計算模式相結合，使計算程序與數據存儲盡可能的在同一個虛擬機上，保證數據的極大吞吐量。HDFS采用master/slave架構。HDFS集群由一個Namenode和若干個Datanode組成。Namenode設置于中心服務器，負責管理整個HDFS的名字空間和用戶對HDFS的訪問；Datanode是分布在不同虛擬機上的數據節點，負責用戶對數據的讀寫等操作。HDFS同時提出了數據均衡的方案，系統會自動的將數據從一個容量不足數據節點上轉移到其他空閑節點，保證整個文件系統的數據均衡。HDFS的結構示意圖可參考文獻 ［15］。

2.2 并行算法設計

在算法1中，我們知道，ALS算法一次迭代需要分別計算U和V，而求U或求V這個過程是比較耗時的，而這個過程正是算法并行之所在。根據式（4）和（5），我們知道，在計算每一個用戶的特征向量Ui時，與它相關的量只有電影特征矩陣V和該用戶i評過分的電影的集合即Ri，Ri是一個向量；同理在計算每一部電影的特征向量Vj時，與它相關的量只有用戶特征矩陣U和評價過電影j的用戶的集合，即Rj，Rj是一個向量。用戶與用戶之間，電影與電影之間是沒有聯系的，所以我們在計算用戶或電影的特征向量時，是可以通過并行方式來處理的。

基于MapReduce的ALS算法，一次迭代需要啟動兩次MapReduce過程，每次求U或V都需啟動一次MapReduce過程。每次MapReduce過程，執行算法1中的步驟（2）或（3）。Hadoop的MapReduce編程模型有兩個階段Map（映射）和Reduce（規約），由于用戶ID和電影ID的唯一性，基于MapReduce的ALS算法并不需要Reduce過程。

基于MapReduce的ALS算法求解U步驟如下：①輸入用戶評過分的電影的集合R［n］（n為用戶數量）及電影特征矩陣V。②啟動MapReduce過程，將電影特征矩陣V分發到各個節點。輸入為存在DFS上的用戶評過分的電影的集合R ［n］。③Map過程：輸入為R ［1…n］，對于R［i］，利用式（4），計算用戶i的特征向量 Ui。輸出為i，Ui。其中i為key，Ui為value。

同理，基于MapReduce的ALS算法求解V步驟如下：①輸入評價過電影的用戶的集合R［m］（m為電影數量）及用戶特征矩陣U。②啟動MapReduce過程，將用戶特征矩陣U分發到各個節點。輸入為存在DFS上的評價過電影的用戶的集合R ［m］。③Map過程：輸入為R ［1…m］，對于R ［j］，利用式（5），計算電影j的特征向量Vj。輸出為j，Vj。其中j為key，Vj為value。

2.3 并行化算法實現細節

2.3.1 數據預處理

在實驗中使用的原始數據集是由一條一條用戶的評分記錄組成的。每一條評分記錄都以一個三元組表示：（UserID，ItemID，Rate）。其表示的含義是某一個用戶對某一個對象進行打分（這里的評分我們統一用1到5分，1分表示非常不喜歡，5分代表非常喜歡）。我們需要將其進行處理得到：用戶i評過分的電影的集合Ri和評價過電影j的用戶的集合Rj，如果有n個用戶，m部電影，則一共有n個Ri，m個Rj。

對于式（4），Ri.集合中的元素不僅僅只是評分，還需要保存用戶評價過哪些電影。因為分數是1到5分，所以Ri.集合中的元素是ItemID＊10＋Rate。這樣電影與評分一一對應，而不丟失信息。

如：ID為1的用戶評過電影1，3，5，評分記錄為：

則 Ri.表示為：

同理對于式（5），合中的元素不僅僅只是評分，還需要保存電影被哪些用戶評價過。所以R.j集合中的元素是UserID＊10＋Rate。這樣用戶與評分一一對應，而不丟失信息。

如：ID為1的電影被用戶1，2，3評過分，評分記錄為：

則 R.j表示為：

這個數據預處理過程也可以利用Hadoop的 MapReduce實現。這個預處理需要啟動兩次MapReduce，一次求用戶評過分的電影的n個集合Ri.，一次求評價過電影的用戶的m個集合R.j。MapReduce編程模型默認的輸入格式是文本輸入，每一行數據都是一條記錄。Map函數接受一組數據并將其轉換為一個key/value對列表，輸入域中的每個元素對應一個key/value對。Reduce函數接受 Map函數生成的列表，然后根據它們的鍵（為每個鍵生成一個鍵/值對）縮小key/value對列表。

如以下是4條評分記錄：

在求用戶評過分的電影的集合時，運行Map函數將得出以下的key/value對列表：

如果對這個key/value對列表應用Reduce函數，將得到以下一組key/value對：

同理在求評價過電影的用戶的集合，運行Map函數將得出以下以下的key/value對列表：

如果對這個key/value對列表應用Reduce函數，將得到以下一組key/value對：

數據預處理中MapReduce的邏輯數據流如圖1所示。

圖1 MapReduce的邏輯數據流

2.3.2 Hadoop參數傳遞

在ALS算法中，求用戶特征矩陣U時需要事先知道電影特征矩陣V，求電影特征矩陣V時需要事先知道用戶特征矩陣U，所以求U時，需要將V作為參數傳遞到算U的函數中；求V時，需要將U作為參數傳遞給算V的函數中。將ALS算法實現在單節點時，只需要將U和V設置成全局變量，就可以解決參數傳遞問題。

然而在基于MapReduce的ALS算法中，我們知道在將MapReduce作業提交給Hadoop集群時，相關的輸入數據將按照Block的大小首先被劃分為多個片，分發到各個節點進行計算，各個節點在計算時只執行MapReduce任務，所以MapReduce編程模型并不支持全局變量。然而在求用戶特征矩陣U時，每個節點計算過程都需要知道V，這時，我們就需要將V作為參數傳遞到各個節點。選擇合適的方式來傳遞參數既能提高工作效率，也可以避免bug的產生。

在基于MapReduce的ALS算法中，求用戶特征矩陣U時，電影特征矩陣V是存在DFS中。我們知道在MapReduce過程中，會將輸入數據按照Block的大小分塊，假設分成批p塊，則有p個Map任務，每個Map任務求解K個用戶特征向量。如果在求每個用戶的特征向量Ui時，都從DFS中讀取電影特征矩陣V，那樣效率會非常低，因為如果有n個用戶，則需從DFS中讀取n次。并且V矩陣一般比較大，其大小是m×d（m是電影數量，d為特征數），所以這種參數傳遞方式效率非常低，并不可取。

同理求電影特征矩陣V時，這種參數傳遞方式也不可取。每個Map任務在開始前都會進行初始化操作，如果在初始化時，讀取文件放到變量中，將這個變量做為整個Map任務的共享變量，則讀取文件次數將減少，有多少個Map任務，只需要讀多少次文件，效率將大大提高。

Hadoop的分布式緩存機制使得一個job的所有map或reduce可以訪問同一份文件。在任務提交后，hadoop將由－files和－archive選項指定的文件復制到 HDFS上（Job－Tracker的文件系統）。在任務運行前，TaskTracker從Job－Tracker文件系統復制文件到本地磁盤作為緩存，這樣任務就可以訪問這些文件。對于job來說，它并不關心文件是從哪兒來的。在使用hadoop的緩存文件DistributedCache時，對于本地化文件的訪問，通常使用Symbolic Link來訪問，這樣更方便。通過 URI hdfs：//namenode/test/input/file1＃myfile指定的文件在當前工作目錄中被符號鏈接為myfile。這樣job里面可直接通過myfile來訪問文件，而不用關心該文件在本地的具體路徑。

2.3.3 Map函數實現

由于用戶ID和電影ID的唯一性，基于MapReduce的ALS算法并不需要Reduce過程。Map函數主要對輸入的每一條數據進行處理，其默認的輸入格式是文本輸入，每一行數據都是一條記錄。在數據預處理時，我們已經知道，每一行的數據格式是：

或者：

Map的任務就是求每一個用戶或每一個對象的特征向量。

3 實驗評估

本章主要對ALS算法在Hadoop平臺實現的性能進行評估，并闡述實驗環境和實驗結果。

3.1 Hadoop集群配置

我們的Hadoop集群配置如下：在實驗中分別使用了一臺Master、2臺Slave和一臺Master、5臺Slave組成的Hadoop集群。所有的機器都是HP計算機，每臺計算機配置為4顆Intel（R）Core（TM）i7處理器，8GB內存。這些機器都處于同一個局域網內。

3.2 實驗數據集

在這個實驗中，我們使用的數據集是Netflix對外發布的一個電影評分數據集［2，5］。這個數據集包括了480 189個用戶在對17 770部電影的103，297，638個評分。所有的評分值都是1到5中的整數值，其中分數越高表示客戶對相應電影的評價越高（越喜歡）。這個數據集非常稀疏，有將近99%的評分值未知。從這個數據集中隨機抽取140萬條評分記錄作為測試集TestSet，其余作為訓練集TrainSet。

3.3 實驗結果

本論文進行了3個實驗，分別是ALS算法在單節點的實現，在一臺Master、2臺Slave的Hadoop集群中的實現和在一臺Master、5臺Slave的Hadoop集群中的實現。

ALS算法中有兩個參數，分別是特征個數和迭代次數。

在第一輪實驗中，設定迭代次數為1，特征個數分別為10，20，30，40，50。最終實驗結果如圖2所示。

圖2 ALS迭代次數為1的實驗結果

橫坐標為特征個數，縱坐標為時間，單位為秒（S）。

由圖2可以看出，隨著特征數的增多，ALS算法在單節點運行的時間與在Hadoop集群上運行的時間之間的比例是越來越大，當特征數為50時，其比例達到了5。由式（4）和式（5）可知，特征數的增多，意味著運算復雜度增加。當特征數為10時，ALS算法在單節點運行的時間比在Hadoop集群運行的要快，這是因為Hadoop集群進行并行化時，master需要進行調度，特征數為10時，運算復雜度并不高，所以用Hadoop集群進行并行計算，并不能體現出并行計算的威力。

Hadoop集群中節點的增多，意味著其運算能力的增加，從圖2可以看出，5個nodes的Hadoop集群運算效率要比2個nodes的Hadoop集群高。當然，并不是越多機器越好，假設每個節點處理一個Map任務，當節點數多于Map任務時，節點的增多并不能提高運算效率，此時多于的機器會被閑置。所以理想情況下，Map任務數最好是Hadoop集群節點數的倍數，這樣才能有效充分利用Hadoop集群的運算能力。

在第二輪實驗中，設定迭代次數為10，特征個數分別為10，20，30，40，50。最終實驗結果如圖3所示。

圖3 ALS迭代次數為10的實驗結果

橫坐標為特征個數，縱坐標為時間，單位為：分鐘。

在這兩個實驗中，可以看出特征數越多，迭代次數越多，ALS算法在Hadoop集群上的運算效率會提高的越多。

4 結束語

本文對推薦系統的協同過濾算法進行介紹，并針對其中基于矩陣分解的ALS算法進行了詳細介紹；同時還對Hadoop平臺產生的背景，應用背景，平臺架構和核心部分做了比較詳細的介紹；然后在上述基礎上實現ALS算法在Hadoop平臺的并行化，以提高算法性能。

通過實驗，我們可以清楚看到基于Hadoop平臺實現的算法在運算效率上提高的非常明顯。當然，在實驗中，我們的實驗數據集還不夠大，還不能完全體現出Hadoop的優勢來，據我們所知Yahoo為了滿足廣告系統和web搜索的研究，在4000個服務器集群上部署Hadoop；Facebook使用了1000個節點的集群部署HDFS，以支持其大量的日志數據的存儲；淘寶網則使用hadoop集群網絡處理大量的電子商務相關的數據；百度公司利用hadoop并行計算系統，進行大規模網頁的分析與搜索，其處理數據達每周200TB。因此協同過濾推薦算法的并行化至關重要，其應用前景非常廣泛。

在本實驗中我們實現了基于矩陣分解的ALS協同過濾算法在Hadoop平臺的并行化，在時間效率上有提高，但Hadoop集群除了受集群機器數影響，還有受一些參數配置的影響，如：文件Block的大小，文件的復制數等，這些參數配置對Hadoop集群的影響將是下一步工作。本文所提出的并行化ALS的算法思想還可以運用于并行化其他的協調過濾算法，如RBM、NNMF等。

［1］LUO Xin，OU YANG Yuanxin，XIONG Zhang，et al.The effect of similarity support in K－nearest－neighborhood based collaborati ve filtering ［J］.Chinese Journal of Computers，2010，33（8）：1437－1445（in Chinese）.［羅辛，歐陽元新，熊璋，等.通過相似度支持度優化基于K近鄰的協同過濾算法 ［J］.計算機學報，2010，33（8）：1437－1445.］

［2］Ricci F，Rokach L，Shapira B，et al.Recommender system handbook ［M］.New York：Springer，2011：1－29.

［3］Das A，Datar M，Garg A，et al.Google news personalization：Scalable online collaborative filtering ［C］.Canada：Proceedings of the 16th International Conference on World Wide Web，2007：271－280.

［4］Adomavicius G，Tuzhilin A.Toward the next generation of recommender systems：A survey of the state－of－the－art and possible extenstions ［J］.TKDE，2005，17（6）：734－749.

［5］ WU J L.Collaborative filtering on the netflix prize dataset［EB/OL］.［2010－08－01］.http：//dsec.pku.edu.cn/～jinlong/

［6］PAN R，ZHOU Y，CAO B，et al.One－class collaborative filtering［C］.Pisa，Italy：Proceedings of the Eighth IEEE International Conference on Data Mining，2008：502－511.

［7］PAN R，Martin S.Mind the gaps：Weighting the unknown in large－scale one－class collaborative filtering ［C］.Paris，France：Proceedings of the 15th ACM SIGKDD International Conference on Knowledge Discovery and Data Mining，2009：667－676.

［8］ZHOU Y H，Wilkinson D，Schreiber R，et al.Large－scale parallel collaborative filtering for the Netflix prize ［C］.Berlin：Proceedings of the 4th International Conference on Algorithmic Aspects in Information and Management，2008：337－348.

［9］Srebro N，Rennie J D M，Jaakkola T.Maximum－margin matrix factorization ［C］.Vancouver：MIT Press（NIPS），2004：1329－1336.

［10］Rennie J D M，Srebro N.Fast maximum margin matrix factorization for collaborative prediction ［C］.Bonn，Germany：Proceedings of the 22nd International Conference on Machine Learning，2005：713－719.

［11］Salakhutdinov R，Mnih A.Probabilistic matrix factorization ［C］.Vancouver，British Columbia，Canada：Proceedings of the 25th International Conference on Machine Learning，2007：1257－1264.

［12］Salakhutdinov R，Mnih A，Hinton G.Restricted boltzmann machines for collaborative filtering ［C］.NY，USA：Proceedings of the 24th International Conference on Machine Learning，2007：791－798.

［13］LEE D D，Seung H S.Learning the parts of objects by non－negative matrix factorization ［J］.Nature，1999，401（11）：788－791.

［14］Tom Wbite.Hadoop：The definitive guide ［M］.2nd ed.USA：O＇Reilly Media，Inc，2010：1－60.

［15］Apach HDFS Architecture ［EB/OL］.http：//hadoop.apache.org/hdfs/docs/current/cn/hdfs＿design.html.

［16］Jeffrey Dean，Sanjay Ghemawat.Map reduce：Simplified data processing on large clusters ［C］.San Francisco，CA：Proceedings of the 6th Conference on Symposium on Opearting Systems Design ＆Implementation，2004：137－150.