Spark的协同过滤.Vs.Hadoop MR
基于物品的協(xié)同過濾推薦算法案例在TDW Spark與MapReudce上的實現(xiàn)對比，相比于MapReduce，TDW Spark執(zhí)行時間減少了66%，計算成本降低了40%。
原文鏈接：http://www.csdn.net/article/2014-11-04/2822474
算法介紹
互聯(lián)網(wǎng)的發(fā)展導(dǎo)致了信息爆炸。面對海量的信息，如何對信息進(jìn)行刷選和過濾，將用戶最關(guān)注最感興趣的信息展現(xiàn)在用戶面前，已經(jīng)成為了一個亟待解決的問題。推薦系統(tǒng)可以通過用戶與信息之間的聯(lián)系，一方面幫助用戶獲取有用的信息，另一方面又能讓信息展現(xiàn)在對其感興趣的用戶面前，實現(xiàn)了信息提供商與用戶的雙贏。
協(xié)同過濾推薦（Collaborative Filtering Recommendation）算法是最經(jīng)典最常用的推薦算法，算法通過分析用戶興趣，在用戶群中找到指定用戶的相似用戶，綜合這些相似用戶對某一信息的評價，形成系統(tǒng)對該指定用戶對此信息的喜好程度預(yù)測。協(xié)同過濾可細(xì)分為以下三種：
- User-based CF: 基于User的協(xié)同過濾，通過不同用戶對Item的評分來評測用戶之間的相似性，根據(jù)用戶之間的相似性做出推薦；
- Item-based CF: 基于Item的協(xié)同過濾，通過用戶對不同Item的評分來評測Item之間的相似性，根據(jù)Item之間的相似性做出推薦；
- Model-based CF: 以模型為基礎(chǔ)的協(xié)同過濾（Model-basedCollaborative Filtering）是先用歷史資料得到一個模型，再用此模型進(jìn)行預(yù)測推薦。
問題描述
輸入數(shù)據(jù)格式：Uid，ItemId，Rating?（用戶Uid對ItemId的評分）。
輸出數(shù)據(jù)：每個ItemId相似性最高的前N個ItemId。
由于篇幅限制，這里我們只選擇基于Item的協(xié)同過濾算法解決這個例子。
算法邏輯
基于Item的協(xié)同過濾算法的基本假設(shè)為兩個相似的Item獲得同一個用戶的好評的可能性較高。因此，該算法首先計算用戶對物品的喜好程度，然后根據(jù)用戶的喜好計算Item之間的相似度，最后找出與每個Item最相似的前N個Item。該算法的詳細(xì)描述如下：
- 計算用戶喜好：不同用戶對Item的評分?jǐn)?shù)值可能相差較大，因此需要先對每個用戶的評分做二元化處理，例如對于某一用戶對某一Item的評分大于其給出的平均評分則標(biāo)記為好評1，否則為差評0。
- 計算Item相似性：采用Jaccard系數(shù)作為計算兩個Item的相似性方法。狹義Jaccard相似度適合計算兩個集合之間的相似程度，計算方法為兩個集合的交集除以其并集，具體的分為以下三步。
2)Item好評鍵值對統(tǒng)計，統(tǒng)計任意兩個有關(guān)聯(lián)Item的相同好評用戶數(shù)。
3)Item相似性計算，計算任意兩個有關(guān)聯(lián)Item的相似度。
- 找出最相似的前N個Item。這一步中，Item的相似度還需要歸一化后整合，然后求出每個Item最相似的前N個Item，具體的分為以下三步。
2)Item相似性評分整合。
3)獲取每個Item相似性最高的前N個Item。
基于MapReduce的實現(xiàn)方案
使用 MapReduce編程模型需要為每一步實現(xiàn)一個MapReduce作業(yè)，一共存在包含七個MapRduce作業(yè)。每個MapReduce作業(yè)都包含Map和Reduce，其中Map從HDFS讀取數(shù)，輸出數(shù)據(jù)通過Shuffle把鍵值對發(fā)送到Reduce，Reduce階段以<key，Iterator<value>>作為輸入，輸出經(jīng)過處理的鍵值對到HDFS。其運行原理如圖1 所示。
圖1
七個MapReduce作業(yè)意味著需要七次讀取和寫入HDFS，而它們的輸入輸出數(shù)據(jù)存在關(guān)聯(lián)，七個作業(yè)輸入輸出數(shù)據(jù)關(guān)系如圖2所示。
圖2
基于MapReduce實現(xiàn)此算法存在以下問題：
- 為了實現(xiàn)一個業(yè)務(wù)邏輯需要使用七個MapReduce作業(yè)，七個作業(yè)間的數(shù)據(jù)交換通過HDFS完成，增加了網(wǎng)絡(luò)和磁盤的開銷。
- 七個作業(yè)都需要分別調(diào)度到集群中運行，增加了Gaia集群的資源調(diào)度開銷。
- MR2和MR3重復(fù)讀取相同的數(shù)據(jù)，造成冗余的HDFS讀寫開銷。
基于Spark的實現(xiàn)方案
相比與MapReduce編程模型，Spark提供了更加靈活的DAG（Directed Acyclic Graph） 編程模型， 不僅包含傳統(tǒng)的map、reduce接口， 還增加了filter、flatMap、union等操作接口，使得編寫Spark程序更加靈活方便。使用Spark編程接口實現(xiàn)上述的業(yè)務(wù)邏輯如圖3所示。
圖3
相對于MapReduce，Spark在以下方面優(yōu)化了作業(yè)的執(zhí)行時間和資源使用。
- DAG編程模型。 通過Spark的DAG編程模型可以把七個MapReduce簡化為一個Spark作業(yè)。Spark會把該作業(yè)自動切分為八個Stage，每個Stage包含多個可并行執(zhí)行的Tasks。Stage之間的數(shù)據(jù)通過Shuffle傳遞。最終只需要讀取和寫入HDFS一次。減少了六次HDFS的讀寫，讀寫HDFS減少了70%。
- Spark作業(yè)啟動后會申請所需的Executor資源，所有Stage的Tasks以線程的方式運行，共用Executors，相對于MapReduce方式，Spark申請資源的次數(shù)減少了近90%。
- Spark引入了RDD（ResilientDistributed Dataset）模型，中間數(shù)據(jù)都以RDD的形式存儲，而RDD分布存儲于slave節(jié)點的內(nèi)存中，這就減少了計算過程中讀寫磁盤的次數(shù)。RDD還提供了Cache機(jī)制，例如對上圖的rdd3進(jìn)行Cache后，rdd4和rdd7都可以訪問rdd3的數(shù)據(jù)。相對于MapReduce減少MR2和MR3重復(fù)讀取相同數(shù)據(jù)的問題。
效果對比
測試使用相同規(guī)模的資源，其中MapReduce方式包含200個Map和100個Reduce，每個Map和Reduce配置4G的內(nèi)存； 由于Spark不再需要Reduce資源， 而MapReduce主要邏輯和資源消耗在Map端，因此使用200和400個Executor做測試，每個Executor包含4G內(nèi)存。測試結(jié)果如下表所示，其中輸入記錄約38億條。| 運行模式 | 計算資源 | 運行時間（min） | 成本（Slot*秒） |
| MapReduce | 200 Map+100 Reduce（4G） | 120 | 693872 |
| Spark | 200 Executor（4G） | 33 | 396000 |
| Spark | 400 Executor（4G） | 21 | 504000 |
對比結(jié)果表的第一行和第二行，Spark運行效率和成本相對于MapReduce方式減少非常明顯，其中，DAG模型減少了70%的HDFS讀寫、cache減少重復(fù)數(shù)據(jù)的讀取，這兩個優(yōu)化即能減少作業(yè)運行時間又能降低成本；而資源調(diào)度次數(shù)的減少能提高作業(yè)的運行效率。
對比結(jié)果表的第二行和第三行，增加一倍的Executor數(shù)目，作業(yè)運行時間減少約50%，成本增加約25%，從這個結(jié)果看到，增加Executor資源能有效的減少作業(yè)的運行時間，但并沒有做到完全線性增加。這是因為每個Task的運行時間并不是完全相等的， 例如某些task處理的數(shù)據(jù)量比其他task多；這可能導(dǎo)致Stage的最后時刻某些Task未結(jié)束而無法啟動下一個Stage，另一方面作業(yè)是一直占有Executor的，這時候會出現(xiàn)一些Executor空閑的狀況，于是導(dǎo)致了成本的增加。
小結(jié)
數(shù)據(jù)挖掘類業(yè)務(wù)大多具有復(fù)雜的處理邏輯，傳統(tǒng)的MapReduce／Pig類框架在應(yīng)對此類數(shù)據(jù)處理任務(wù)時存在著嚴(yán)重的性能問題。針對這些任務(wù)，如果利用Spark的迭代計算和內(nèi)存計算優(yōu)勢，將會大幅降低運行時間和計算成本。TDW目前已經(jīng)維護(hù)了千臺規(guī)模的Spark集群，并且會在資源利用率、穩(wěn)定性和易用性等方面做進(jìn)一步的提升和改進(jìn)，為業(yè)務(wù)提供更有利的支持。
總結(jié)
以上是生活随笔為你收集整理的Spark的协同过滤.Vs.Hadoop MR的全部內(nèi)容，希望文章能夠幫你解決所遇到的問題。
- 上一篇： 互联网的大数据神话——NoSQL
- 下一篇： 基于物品的协同过滤ItemCF的mapr