Spark vs. MapReduce 時(shí)間節(jié)約66%，計(jì)算節(jié)約40%

MapReduce為大數(shù)據(jù)挖掘提供了有力的支持，但是復(fù)雜的挖掘算法往往需要多個(gè)MapReduce作業(yè)才能完成，多個(gè)作業(yè)之間存在著冗余的磁盤(pán)讀寫(xiě)開(kāi)銷和多次資源申請(qǐng)過(guò)程，使得基于MapReduce的算法實(shí)現(xiàn)存在嚴(yán)重的性能問(wèn)題。后起之秀Spark得益于其在迭代計(jì)算和內(nèi)存計(jì)算上的優(yōu)勢(shì)，可以自動(dòng)調(diào)度復(fù)雜的計(jì)算任務(wù)，避免中間結(jié)果的磁盤(pán)讀寫(xiě)和資源申請(qǐng)過(guò)程，非常適合數(shù)據(jù)挖掘算法。騰訊TDW Spark平臺(tái)基于社區(qū)最新Spark版本進(jìn)行深度改造，在性能、穩(wěn)定和規(guī)模方面都得到了極大的提高，為大數(shù)據(jù)挖掘任務(wù)提供了有力的支持。

　　本文將介紹基于物品的協(xié)同過(guò)濾推薦算法案例在TDW Spark與MapReudce上的實(shí)現(xiàn)對(duì)比，相比于MapReduce，TDW Spark執(zhí)行時(shí)間減少了66%，計(jì)算成本降低了40%。

　　算法介紹

　　互聯(lián)網(wǎng)的發(fā)展導(dǎo)致了信息爆炸。面對(duì)海量的信息，如何對(duì)信息進(jìn)行刷選和過(guò)濾，將用戶最關(guān)注最感興趣的信息展現(xiàn)在用戶面前，已經(jīng)成為了一個(gè)亟待解決的問(wèn)題。推薦系統(tǒng)可以通過(guò)用戶與信息之間的聯(lián)系，一方面幫助用戶獲取有用的信息，另一方面又能讓信息展現(xiàn)在對(duì)其感興趣的用戶面前，實(shí)現(xiàn)了信息提供商與用戶的雙贏。

　　協(xié)同過(guò)濾推薦(Collaborative Filtering Recommendation)算法是最經(jīng)典最常用的推薦算法，算法通過(guò)分析用戶興趣，在用戶群中找到指定用戶的相似用戶，綜合這些相似用戶對(duì)某一信息的評(píng)價(jià)，形成系統(tǒng)對(duì)該指定用戶對(duì)此信息的喜好程度預(yù)測(cè)。協(xié)同過(guò)濾可細(xì)分為以下三種：

　　User-based CF: 基于User的協(xié)同過(guò)濾，通過(guò)不同用戶對(duì)Item的評(píng)分來(lái)評(píng)測(cè)用戶之間的相似性，根據(jù)用戶之間的相似性做出推薦;

　　Item-based CF: 基于Item的協(xié)同過(guò)濾，通過(guò)用戶對(duì)不同Item的評(píng)分來(lái)評(píng)測(cè)Item之間的相似性，根據(jù)Item之間的相似性做出推薦;

　　Model-based CF: 以模型為基礎(chǔ)的協(xié)同過(guò)濾(Model-based Collaborative Filtering)是先用歷史資料得到一個(gè)模型，再用此模型進(jìn)行預(yù)測(cè)推薦。

　　問(wèn)題描述

　　輸入數(shù)據(jù)格式：Uid，ItemId，Rating (用戶Uid對(duì)ItemId的評(píng)分)。

　　輸出數(shù)據(jù)：每個(gè)ItemId相似性最高的前N個(gè)ItemId。

　　由于篇幅限制，這里我們只選擇基于Item的協(xié)同過(guò)濾算法解決這個(gè)例子。

　　算法邏輯

　　基于Item的協(xié)同過(guò)濾算法的基本假設(shè)為兩個(gè)相似的Item獲得同一個(gè)用戶的好評(píng)的可能性較高。因此，該算法首先計(jì)算用戶對(duì)物品的喜好程度，然后根據(jù)用戶的喜好計(jì)算Item之間的相似度，最后找出與每個(gè)Item最相似的前N個(gè)Item。該算法的詳細(xì)描述如下：

　　計(jì)算用戶喜好：不同用戶對(duì)Item的評(píng)分?jǐn)?shù)值可能相差較大，因此需要先對(duì)每個(gè)用戶的評(píng)分做二元化處理，例如對(duì)于某一用戶對(duì)某一Item的評(píng)分大于其給出的平均評(píng)分則標(biāo)記為好評(píng)1，否則為差評(píng)0。

　　計(jì)算Item相似性：采用Jaccard系數(shù)作為計(jì)算兩個(gè)Item的相似性方法。狹義Jaccard相似度適合計(jì)算兩個(gè)集合之間的相似程度，計(jì)算方法為兩個(gè)集合的交集除以其并集，具體的分為以下三步。

　　1) Item好評(píng)數(shù)統(tǒng)計(jì)，統(tǒng)計(jì)每個(gè)Item的好評(píng)用戶數(shù)。

　　2) Item好評(píng)鍵值對(duì)統(tǒng)計(jì)，統(tǒng)計(jì)任意兩個(gè)有關(guān)聯(lián)Item的相同好評(píng)用戶數(shù)。

　　3) Item相似性計(jì)算，計(jì)算任意兩個(gè)有關(guān)聯(lián)Item的相似度。

　　找出最相似的前N個(gè)Item。這一步中，Item的相似度還需要?dú)w一化后整合，然后求出每個(gè)Item最相似的前N個(gè)Item，具體的分為以下三步。

　　1) Item相似性歸一化。

　　2) Item相似性評(píng)分整合。

　　3) 獲取每個(gè)Item相似性最高的前N個(gè)Item。

　　基于MapReduce的實(shí)現(xiàn)方案

　　使用MapReduce編程模型需要為每一步實(shí)現(xiàn)一個(gè)MapReduce作業(yè)，一共存在包含七個(gè)MapRduce作業(yè)。每個(gè)MapReduce作業(yè)都包含Map和Reduce，其中Map從HDFS讀取數(shù)，輸出數(shù)據(jù)通過(guò)Shuffle把鍵值對(duì)發(fā)送到Reduce，Reduce階段以<key，iterator>作為輸入，輸出經(jīng)過(guò)處理的鍵值對(duì)到HDFS。其運(yùn)行原理如圖1 所示。

　　七個(gè)MapReduce作業(yè)意味著需要七次讀取和寫(xiě)入HDFS，而它們的輸入輸出數(shù)據(jù)存在關(guān)聯(lián)，七個(gè)作業(yè)輸入輸出數(shù)據(jù)關(guān)系如圖2所示。

　　基于MapReduce實(shí)現(xiàn)此算法存在以下問(wèn)題：

　　為了實(shí)現(xiàn)一個(gè)業(yè)務(wù)邏輯需要使用七個(gè)MapReduce作業(yè)，七個(gè)作業(yè)間的數(shù)據(jù)交換通過(guò)HDFS完成，增加了網(wǎng)絡(luò)和磁盤(pán)的開(kāi)銷。

　　七個(gè)作業(yè)都需要分別調(diào)度到集群中運(yùn)行，增加了Gaia集群的資源調(diào)度開(kāi)銷。

　　MR2和MR3重復(fù)讀取相同的數(shù)據(jù)，造成冗余的HDFS讀寫(xiě)開(kāi)銷。

　　這些問(wèn)題導(dǎo)致作業(yè)運(yùn)行時(shí)間大大增長(zhǎng)，作業(yè)成本增加。

　　基于Spark的實(shí)現(xiàn)方案

　　相比與MapReduce編程模型，Spark提供了更加靈活的DAG(Directed Acyclic Graph) 編程模型， 不僅包含傳統(tǒng)的map、reduce接口， 還增加了filter、flatMap、union等操作接口，使得編寫(xiě)Spark程序更加靈活方便。使用Spark編程接口實(shí)現(xiàn)上述的業(yè)務(wù)邏輯如圖3所示。

　　相對(duì)于MapReduce，Spark在以下方面優(yōu)化了作業(yè)的執(zhí)行時(shí)間和資源使用。

　　DAG編程模型。 通過(guò)Spark的DAG編程模型可以把七個(gè)MapReduce簡(jiǎn)化為一個(gè)Spark作業(yè)。Spark會(huì)把該作業(yè)自動(dòng)切分為八個(gè)Stage，每個(gè)Stage包含多個(gè)可并行執(zhí)行的Tasks。Stage之間的數(shù)據(jù)通過(guò)Shuffle傳遞。最終只需要讀取和寫(xiě)入HDFS一次。減少了六次HDFS的讀寫(xiě)，讀寫(xiě)HDFS減少了70%。

　　Spark作業(yè)啟動(dòng)后會(huì)申請(qǐng)所需的Executor資源，所有Stage的Tasks以線程的方式運(yùn)行，共用Executors，相對(duì)于MapReduce方式，Spark申請(qǐng)資源的次數(shù)減少了近90%。

　　Spark引入了RDD(Resilient Distributed Dataset)模型，中間數(shù)據(jù)都以RDD的形式存儲(chǔ)，而RDD分布存儲(chǔ)于slave節(jié)點(diǎn)的內(nèi)存中，這就減少了計(jì)算過(guò)程中讀寫(xiě)磁盤(pán)的次數(shù)。RDD還提供了Cache機(jī)制，例如對(duì)上圖的rdd3進(jìn)行Cache后，rdd4和rdd7都可以訪問(wèn)rdd3的數(shù)據(jù)。相對(duì)于MapReduce減少M(fèi)R2和MR3重復(fù)讀取相同數(shù)據(jù)的問(wèn)題。

　　效果對(duì)比

　　測(cè)試使用相同規(guī)模的資源，其中MapReduce方式包含200個(gè)Map和100個(gè)Reduce，每個(gè)Map和Reduce配置4G的內(nèi)存; 由于Spark不再需要Reduce資源， 而MapReduce主要邏輯和資源消耗在Map端，因此使用200和400個(gè)Executor做測(cè)試，每個(gè)Executor包含4G內(nèi)存。測(cè)試結(jié)果如下表所示，其中輸入記錄約38億條。

　　運(yùn)行模式計(jì)算資源運(yùn)行時(shí)間(min)成本(Slot*秒)

　　MapReduce200 Map+100 Reduce(4G)120693872

　　Spark200 Executor(4G)33396000

　　Spark400 Executor(4G)21504000

　　對(duì)比結(jié)果表的第一行和第二行，Spark運(yùn)行效率和成本相對(duì)于MapReduce方式減少非常明顯，其中，DAG模型減少了70%的HDFS讀寫(xiě)、cache減少重復(fù)數(shù)據(jù)的讀取，這兩個(gè)優(yōu)化即能減少作業(yè)運(yùn)行時(shí)間又能降低成本;而資源調(diào)度次數(shù)的減少能提高作業(yè)的運(yùn)行效率。

　　對(duì)比結(jié)果表的第二行和第三行，增加一倍的Executor數(shù)目，作業(yè)運(yùn)行時(shí)間減少約50%，成本增加約25%，從這個(gè)結(jié)果看到，增加Executor資源能有效的減少作業(yè)的運(yùn)行時(shí)間，但并沒(méi)有做到完全線性增加。這是因?yàn)槊總€(gè)Task的運(yùn)行時(shí)間并不是完全相等的， 例如某些task處理的數(shù)據(jù)量比其他task多;這可能導(dǎo)致Stage的最后時(shí)刻某些Task未結(jié)束而無(wú)法啟動(dòng)下一個(gè)Stage，另一方面作業(yè)是一直占有Executor的，這時(shí)候會(huì)出現(xiàn)一些Executor空閑的狀況，于是導(dǎo)致了成本的增加。CDA數(shù)據(jù)分析師培訓(xùn)官網(wǎng)

　　小結(jié)

　　數(shù)據(jù)挖掘類業(yè)務(wù)大多具有復(fù)雜的處理邏輯，傳統(tǒng)的MapReduce/Pig類框架在應(yīng)對(duì)此類數(shù)據(jù)處理任務(wù)時(shí)存在著嚴(yán)重的性能問(wèn)題。針對(duì)這些任務(wù)，如果利用Spark的迭代計(jì)算和內(nèi)存計(jì)算優(yōu)勢(shì)，將會(huì)大幅降低運(yùn)行時(shí)間和計(jì)算成本。TDW目前已經(jīng)維護(hù)了千臺(tái)規(guī)模的Spark集群，并且會(huì)在資源利用率、穩(wěn)定性和易用性等方面做進(jìn)一步的提升和改進(jìn)，為業(yè)務(wù)提供更有利的支持。