微博基于Spark的機(jī)器學(xué)習(xí)應(yīng)用分析

　　眾所周知，自2015年以來微博的業(yè)務(wù)發(fā)展迅猛。如果根據(jù)內(nèi)容來劃分，微博的業(yè)務(wù)有主信息（Feed）流、熱門微博、微博推送（Push）、反垃圾、微博分發(fā)控制等。每個業(yè)務(wù)都有自己不同的用戶構(gòu)成、業(yè)務(wù)關(guān)注點(diǎn)和數(shù)據(jù)特征。龐大的用戶基數(shù)下，由用戶相互關(guān)注衍生的用戶間關(guān)系，以及用戶千人千面的個性化需求，要求我們用更高、更大規(guī)模的維度去刻畫和描繪用戶。大體量的微博內(nèi)容，也呈現(xiàn)出多樣化、多媒體化的發(fā)展趨勢。

　　一直以來，微博都嘗試通過機(jī)器學(xué)習(xí)來解決業(yè)務(wù)場景中遇到的各種挑戰(zhàn)。本文為新浪微博吳磊在CCTC 2017云計算大會Spark峰會所做分享《基于Spark的大規(guī)模機(jī)器學(xué)習(xí)在微博的應(yīng)用》主題的一部分，介紹微博在面對大規(guī)模機(jī)器學(xué)習(xí)的挑戰(zhàn)時，采取的最佳實(shí)踐和解決方案。

　　Spark Mllib

　　針對微博近百億特征維度、近萬億樣本量的模型訓(xùn)練需求，我們首先嘗試了Apache Spark原生實(shí)現(xiàn)的邏輯回歸算法。采用該方式的優(yōu)點(diǎn)顯而易見，即開發(fā)周期短、試錯成本低。我們將不同來源的特征（用戶、微博內(nèi)容、用戶間關(guān)系、使用環(huán)境等）根據(jù)業(yè)務(wù)需要進(jìn)行數(shù)據(jù)清洗、提取、離散化，生成Libsvm格式的可訓(xùn)練樣本集，再將樣本喂給LR算法進(jìn)行訓(xùn)練。在維度升高的過程中，我們遇到了不同方面的問題，并通過實(shí)踐提供了解決辦法。

　　Stack overflow

　　棧溢出的問題在函數(shù)嵌套調(diào)用中非常普遍，但在我們的實(shí)踐中發(fā)現(xiàn)，過多Spark RDD的union操作，同樣會導(dǎo)致棧溢出的問題。解決辦法自然是避免大量的RDD union，轉(zhuǎn)而采用其他的實(shí)現(xiàn)方式。

　　AUC=0.5

　　在進(jìn)行模型訓(xùn)練的過程中，曾出現(xiàn)測試集AUC一直停留在0.5的尷尬局面。通過仔細(xì)查看訓(xùn)練參數(shù)，發(fā)現(xiàn)當(dāng)LR的學(xué)習(xí)率設(shè)置較大時，梯度下降會在局部最優(yōu)左右搖擺，造成訓(xùn)練出來的模型成本偏高，擬合性差。通過適當(dāng)調(diào)整學(xué)習(xí)率可以避免該問題的出現(xiàn)。

　　整型越界

　　整型越界通常是指給定的數(shù)據(jù)值過大，超出了整形（32bit Int）的上限。但在我們的場景中，導(dǎo)致整型越界的并不是某個具體數(shù)據(jù)值的大小，而是因?yàn)橛?xùn)練樣本數(shù)據(jù)量過大、HDFS的分片過大，導(dǎo)致Spark RDD的單個分片內(nèi)的數(shù)據(jù)記錄條數(shù)超出了整型上限，進(jìn)而導(dǎo)致越界。Spark RDD中的迭代器以整數(shù)（Int）來記錄Iterator的位置，當(dāng)記錄數(shù)超過32位整型所包含的范圍（2147483647），就會報出該錯誤。

　　解決辦法是在Spark加載HDFS中的HadoopRDD時，設(shè)置分區(qū)數(shù)，將分區(qū)數(shù)設(shè)置足夠大，從而保證每個分片的數(shù)據(jù)量足夠小，以避免該問題。可以通過公式（總記錄數(shù)／單個分片記錄數(shù)）來計算合理的分區(qū)數(shù)。

　　Shuffle fetch failed

　　在分布式計算中，Shuffle階段不可避免，在Shuffle的Map階段，Spark會將Map輸出緩存到本機(jī)的本地文件系統(tǒng)。當(dāng)Map輸出的數(shù)據(jù)較大，且本地文件系統(tǒng)存儲空間不足時，會導(dǎo)致Shuffle中間文件的丟失，這是Shuffle fetch failed錯誤的常見原因。但在我們的場景中，我們手工設(shè)置了spark.local.dir配置項(xiàng)，將其指向存儲空間足夠、I/O效率較高的文件系統(tǒng)中，但還是碰到了該問題。

　　通過仔細(xì)查對日志和Spark UI的記錄，發(fā)現(xiàn)有個別Executor因任務(wù)過重、GC時間過長，丟失了與Driver的心跳。Driver感知不到這些Executor的心跳，便主動要求Yarn的Application master將包含這些Executor的Container殺掉。

　　皮之不存、毛之焉附，Executor被殺掉了，存儲在其中的Map輸出信息自然也就丟了，造成在Reduce階段，Reducer無法獲得屬于自己的那份Map輸出。解決辦法是合理地設(shè)置JVM的GC設(shè)置，或者通過將spark.network.timeout的時間（默認(rèn)60s）設(shè)置為120s，該時間為Driver與Executor心跳通信的超時時間，給Executor足夠的響應(yīng)時間，讓其不必因處理任務(wù)過重而無暇與Driver端通信。

　　通過各種優(yōu)化，我們將模型的維度提升至千萬維。當(dāng)模型維度沖擊到億維時，因Spark Mllib LR的實(shí)現(xiàn)為非模型并行，過高的模型維度會導(dǎo)致海森矩陣呈指數(shù)級上漲，導(dǎo)致內(nèi)存和網(wǎng)絡(luò)I/O的極大開銷。因此我們不得不嘗試其他的解決方案。

　　基于Spark的參數(shù)服務(wù)器

　　在經(jīng)過大量調(diào)研和初步的嘗試，我們最終選擇參數(shù)服務(wù)器方案來解決模型并行問題。參數(shù)服務(wù)器通過將參數(shù)分片以分布式形式存儲和訪問，將高維模型平均分配到參數(shù)服務(wù)器集群中的每一臺機(jī)器，將CPU計算、內(nèi)存消耗、存儲、磁盤I/O、網(wǎng)絡(luò)I/O等負(fù)載和開銷均攤。典型的參數(shù)服務(wù)器采用主從架構(gòu)，Master負(fù)責(zé)記錄和維護(hù)每個參數(shù)服務(wù)器的心跳和狀態(tài)；參數(shù)服務(wù)器則負(fù)責(zé)參數(shù)分片的存儲、梯度計算、梯度更新、副本存儲等具體工作。圖1是我們采用的參數(shù)服務(wù)器方案。

　　圖1 微博參數(shù)服務(wù)器架構(gòu)圖

　　藍(lán)色文本框架即是采用主從架構(gòu)的參數(shù)服務(wù)器集群，以Yarn應(yīng)用的方式部署在Yarn集群中，為所有應(yīng)用提供服務(wù)。在參數(shù)服務(wù)器的客戶端，也是通過Yarn應(yīng)用的方式，啟動Spark任務(wù)執(zhí)行LR分布式算法。在圖中綠色文本框中，Spark模型訓(xùn)練以獨(dú)立的應(yīng)用存在于Yarn集群中。在模型訓(xùn)練過程中，每個Spark Executor以數(shù)據(jù)分片為單位，進(jìn)行參數(shù)的拉取、計算、更新和推送。

　　在參數(shù)服務(wù)器實(shí)現(xiàn)方面，業(yè)界至少有兩種實(shí)現(xiàn)方式，即全同步與全異步。全同步的方式能夠在理論層面保證模型收斂，但在分布式環(huán)境中，鑒于各計算節(jié)點(diǎn)的執(zhí)行性能各異，加上迭代中需要彼此間相互同步，容易導(dǎo)致過早執(zhí)行完任務(wù)的節(jié)點(diǎn)等待計算任務(wù)繁重的節(jié)點(diǎn)，引入通信邊界，從而造成計算資料的浪費(fèi)和開銷。全異步方式能夠很好地避免這些問題，因節(jié)點(diǎn)間無需等待和同步，可以充分利用各個節(jié)點(diǎn)的計算資源。雖然從理論上無法驗(yàn)證模型一定收斂，但是通過實(shí)踐發(fā)現(xiàn)，模型每次的迭代速度會更快，AUC的加速度會更高，實(shí)際訓(xùn)練出的模型效果可以滿足業(yè)務(wù)和線上的要求。

　　在通過參數(shù)服務(wù)器進(jìn)行LR模型訓(xùn)練時，我們總結(jié)了影響執(zhí)行性能的關(guān)鍵因素，羅列如下：

　　Batch size

　　即Spark數(shù)據(jù)分片大小。前文提到，每個Spark Executor以數(shù)據(jù)分片為單位，進(jìn)行參數(shù)的拉取和推送。分片的大小直接決定本次迭代需要拉取和通信的參數(shù)數(shù)量，而參數(shù)數(shù)量直接決定了本地迭代的計算量、通信量。因此分片大小是影響模型訓(xùn)練執(zhí)行性能的首要因素。過大的數(shù)據(jù)分片會造成單次迭代任務(wù)過重，Executor不堪重負(fù)；過小的分片雖然能夠充分利用網(wǎng)絡(luò)吞吐，但是會造成很多額外的開銷。因此，選擇合理的Batch size，將會令執(zhí)行性能的提升事半功倍。下文將以Batch size為例，對比不同設(shè)置下模型訓(xùn)練執(zhí)行性能的差異。

　　PS server數(shù)量

　　參數(shù)服務(wù)器的數(shù)量，決定了模型參數(shù)的存儲容量。通過擴(kuò)展參數(shù)服務(wù)器集群，理論上可以無限擴(kuò)展存儲容量。但是當(dāng)集群大小達(dá)到瓶頸值時，過多的參數(shù)服務(wù)器帶來的網(wǎng)絡(luò)開銷反而會令整體執(zhí)行性能趨于平緩甚至下降。

　　特征稀疏度

　　根據(jù)需要可以將原始業(yè)務(wù)特征（用戶、微博內(nèi)容、用戶間關(guān)系、使用環(huán)境等）通過映射函數(shù)映射到高維模型，以這種方式提煉出區(qū)分度更佳的特征。特征稀疏度結(jié)合每次迭代數(shù)據(jù)分片的數(shù)據(jù)分布，決定了該分片本次迭代需要拉取和推送的參數(shù)數(shù)量，進(jìn)而決定了本次迭代所需的計算資源和網(wǎng)絡(luò)開銷。

　　PS分區(qū)策略

　　分區(qū)策略決定了模型參數(shù)在參數(shù)服務(wù)器的分布，好的分區(qū)策略能夠使模型參數(shù)的分布更均勻，從而均攤每個節(jié)點(diǎn)的計算和通信負(fù)載。

　　Spark內(nèi)存規(guī)劃

　　在PS的客戶端，Spark Executor需要保證有足夠的內(nèi)存容納本次迭代分片所需的參數(shù)向量，才能完成后續(xù)的參數(shù)計算、更新任務(wù)。

　　下表為不同的Batch size下，各執(zhí)行性能指標(biāo)對比。Parameter（MB）表示一次迭代所需參數(shù)個數(shù)；Tx（MB）表示一次迭代的網(wǎng)絡(luò)吞吐；Pull（ms）和Push（ms）分別表示一次迭代的拉取和推送時間消耗；Time（s）為一次迭代的整體執(zhí)行時間。從表1中可見，參數(shù)個數(shù)與分片大小成正比、網(wǎng)絡(luò)吞吐與分片大小成反比。分片越小，需要通信、處理的參數(shù)越少，但PS客戶端與PS服務(wù)器通信更加頻繁，因而網(wǎng)絡(luò)吞吐更高。但是當(dāng)分片過小時，會產(chǎn)生額外的開銷，造成參數(shù)拉取、推送的平均耗時和任務(wù)的整體耗時上升。

　　表1 模型訓(xùn)練執(zhí)行性能指標(biāo)在不同Batch size下的對比

　　通過參數(shù)服務(wù)器的解決方案，我們解決了微博機(jī)器學(xué)習(xí)平臺化進(jìn)程中的大規(guī)模模型訓(xùn)練問題。眾所周知，在機(jī)器學(xué)習(xí)流中，模型訓(xùn)練只是其中耗時最短的一環(huán)。如果把機(jī)器學(xué)習(xí)流比作烹飪，那么模型訓(xùn)練就是最后翻炒的過程，烹飪的大部分時間實(shí)際上都花在了食材、佐料的挑選，洗菜、擇菜，食材再加工（切丁、切塊、過油、預(yù)熱）等步驟。

　　在微博的機(jī)器學(xué)習(xí)流中，原始樣本生成、數(shù)據(jù)處理、特征工程、訓(xùn)練樣本生成、模型后期的測試、評估等步驟所需要投入的時間和精力，占據(jù)了整個流程的80%之多。如何能夠高效地端到端進(jìn)行機(jī)器學(xué)習(xí)流的開發(fā)，如何能夠根據(jù)線上的反饋及時地選取高區(qū)分度特征，對模型進(jìn)行優(yōu)化，驗(yàn)證模型的有效性，加速模型迭代效率，滿足線上的要求，都是我們需要解決的問題。在新一期《程序員》“weiflow——微博機(jī)器學(xué)習(xí)流統(tǒng)一計算框架”一文中，我們將為你一一解答。