百億特徵維度、近萬億樣本量，微博基於Spark的大規模機器學習應用

作者丨吳磊，張拓宇

編輯丨郭芮

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

一直以來，微博都嘗試通過機器學習來解決業務場景中遇到的各種挑戰。本文為新浪微博吳磊在CCTC 2017雲計算大會Spark峰會所做分享《基於Spark的大規模機器學習在微博的應用》主題的一部分，介紹微博在面對大規模機器學習的挑戰時，採取的最佳實踐和解決方案。

Spark Mllib

針對微博近百億特徵維度、近萬億樣本量的模型訓練需求，我們首先嘗試了Apache Spark原生實現的邏輯回歸演算法。採用該方式的優點顯而易見，即開發周期短、試錯成本低。我們將不同來源的特徵（用戶、微博內容、用戶間關係、使用環境等）根據業務需要進行數據清洗、提取、離散化，生成Libsvm格式的可訓練樣本集，再將樣本餵給LR演算法進行訓練。在維度升高的過程中，我們遇到了不同方面的問題，並通過實踐提供了解決辦法。

Stack overflow

棧溢出的問題在函數嵌套調用中非常普遍，但在我們的實踐中發現，過多Spark RDD的union操作，同樣會導致棧溢出的問題。解決辦法自然是避免大量的RDD union，轉而採用其他的實現方式。

AUC=0.5

在進行模型訓練的過程中，曾出現測試集AUC一直停留在0.5的尷尬局面。通過仔細查看訓練參數，發現當LR的學習率設置較大時，梯度下降會在局部最優左右搖擺，造成訓練出來的模型成本偏高，擬合性差。通過適當調整學習率可以避免該問題的出現。

整型越界

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

Shuffle fetch failed

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

通過各種優化，我們將模型的維度提升至千萬維。當模型維度衝擊到億維時，因Spark Mllib LR的實現為非模型並行，過高的模型維度會導致海森矩陣呈指數級上漲，導致內存和網路I/O的極大開銷。因此我們不得不嘗試其他的解決方案。

基於Spark的參數伺服器

在經過大量調研和初步的嘗試，我們最終選擇參數伺服器方案來解決模型並行問題。參數伺服器通過將參數分片以分布式形式存儲和訪問，將高維模型平均分配到參數伺服器集群中的每一台機器，將CPU計算、內存消耗、存儲、磁碟I/O、網路I/O等負載和開銷均攤。典型的參數伺服器採用主從架構，Master負責記錄和維護每個參數伺服器的心跳和狀態；參數伺服器則負責參數分片的存儲、梯度計算、梯度更新、副本存儲等具體工作。圖1是我們採用的參數伺服器方案。

圖1 微博參數伺服器架構圖

藍色文本框架即是採用主從架構的參數伺服器集群，以Yarn應用的方式部署在Yarn集群中，為所有應用提供服務。在參數伺服器的客戶端，也是通過Yarn應用的方式，啟動Spark任務執行LR分布式演算法。在圖中綠色文本框中，Spark模型訓練以獨立的應用存在於Yarn集群中。在模型訓練過程中，每個Spark Executor以數據分片為單位，進行參數的拉取、計算、更新和推送。

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

在通過參數伺服器進行LR模型訓練時，我們總結了影響執行性能的關鍵因素，羅列如下：

Batch size

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

PS server數量

參數伺服器的數量，決定了模型參數的存儲容量。通過擴展參數伺服器集群，理論上可以無限擴展存儲容量。但是當集群大小達到瓶頸值時，過多的參數伺服器帶來的網路開銷反而會令整體執行性能趨於平緩甚至下降。

特徵稀疏度

根據需要可以將原始業務特徵（用戶、微博內容、用戶間關係、使用環境等）通過映射函數映射到高維模型，以這種方式提煉出區分度更佳的特徵。特徵稀疏度結合每次迭代數據分片的數據分布，決定了該分片本次迭代需要拉取和推送的參數數量，進而決定了本次迭代所需的計算資源和網路開銷。

PS分區策略

分區策略決定了模型參數在參數伺服器的分布，好的分區策略能夠使模型參數的分布更均勻，從而均攤每個節點的計算和通信負載。

Spark內存規劃

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

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

表1 模型訓練執行性能指標在不同Batch size下的對比

通過參數伺服器的解決方案，我們解決了微博機器學習平台化進程中的大規模模型訓練問題。眾所周知，在機器學習流中，模型訓練只是其中耗時最短的一環。如果把機器學習流比作烹飪，那麼模型訓練就是最後翻炒的過程，烹飪的大部分時間實際上都花在了食材、佐料的挑選，洗菜、擇菜，食材再加工（切丁、切塊、過油、預熱）等步驟。

在微博的機器學習流中，原始樣本生成、數據處理、特徵工程、訓練樣本生成、模型後期的測試、評估等步驟所需要投入的時間和精力，佔據了整個流程的80%之多。如何能夠高效地端到端進行機器學習流的開發，如何能夠根據線上的反饋及時地選取高區分度特徵，對模型進行優化，驗證模型的有效性，加速模型迭代效率，滿足線上的要求，都是我們需要解決的問題。在新一期《程序員》「weiflow——微博機器學習流統一計算框架」一文中，我們將為你一一解答。

點擊展開全文

喜歡這篇文章嗎？立刻分享出去讓更多人知道吧！