研究機器學習之MLlib實踐經驗

本文主要討論是用MLlib進行Classification工作。典型的應用場景就是AD CTR Prediction，也就是大部分互聯網公司的利潤來源。據業余了解，廣告CTR預估使用最多的基礎算法還是L1正則化的Logistic Regression。

機器學習任務主要分為兩種：Supervised Machine Learning 和 Unsupervised Machine Learning。其中Supervised Machine Learning主要包括Classification和Regression，Unsupervised Machine Learning主要包括Clustering。除了這些核心的算法以外，還有一些輔助處理的模塊，例如Preprocessing, Dimensionality Reduction, Model Selection等。

目前最新的Spark 1.1.0版本中MLlib主要還是對核心算法的支持，輔助處理模塊還很不完善。源代碼包和其功能的對應關系如下：

本文主要討論是用MLlib進行 Classification工作。分類是機器學習最基礎的工作，典型的應用場景就是AD CTR Prediction，也就是大部分互聯網公司的利潤來源。據業余了解，廣告CTR預估使用最多的基礎算法還是L1正則化的Logistic Regression。

下面一步一步來看看使用MLlib進行Classification機器學習。

1、分類算法原理與MLlib的實現

首先需要了解機器學習和MLlib的基礎知識和原理，大家可以參考 http://spark.apache.org/docs/latest/mllib-Linear-Methods.html 。本文主要從工程實踐的角度討論如何使用和調優。

分類問題主要包括Binary Classification 和 Multiclass Classification。目前的MLlib只支持Linear Classification問題，這里討論的也都是線性分類問題，不涉及到Kernel Method等。目前MLlib里面的Classification算法最常用的就是LR，SVM和Tree/RF。其中LR和SVM目前只支持 Binary Classification，Tree/RF支持Multiclass Classification。本文主要討論使用LR和SVM進行線性Binary Classification問題的實踐中遇到的一些問題。

抽象來看LR和SVM算法都是通過指定Loss Function和Gradient/SUB-Gradient，然后通過Optimization算法（SGD或LBFGS）求使得Loss Function最小的凸優化問題，最后得出的解是一個Weights向量。從代碼中也可以看出，LR和SVM算法僅僅是指定的Loss Function和Gradient是不同的，其求解最小值的過程是通用的，所以求解最小值的過程抽象出了Optimization模塊，目前主要有 SGD和LBFGS兩種實現。

為了防止過擬合，需要在Loss Function后面加入一個正則化項一起求最小值。正則化項相當于對Weights向量的懲罰，期望求出一個更簡單的模型。 MLlib目前支持兩種正則化方法L1和L2。 L2正則化假設模型參數服從高斯分布，L2正則化函數比L1更光滑，所以更容易計算；L1假設模型參數服從拉普拉斯分布，L1正則化具備產生稀疏解的功能，從而具備Feature Selection的能力。

2、兩種Optimization方法SGD和LBFGS

有了上面的數學基礎，現在就是求取一個函數的最小值問題了。MLlib里面目前提供兩種方法SGD和LBFGS。關于這兩種算法的原理，大家可以參考 http://spark.apache.org/docs/latest/mllib-Optimization.html 。這兩種優化方法的核心都是RDD的aggregate操作，這個從Spark Job運行時的UI中可以看出，SGD/LBFGS每迭代一次，aggregate執行一次，Spark UI中出現一個stage。下面分別看看兩種優化算法具體怎么實現的。

SGD：

核心實現在GradientDescent.runMiniBatchSGD函數中

    for (i <- 1 to numIterations) {  
          val bcWeights = data.context.broadcast(weights)  
          // Sample a subset (fraction miniBatchFraction) of the total data  
          // compute and sum up the subgradients on this subset (this is one map-reduce)  
          val (gradientSum, lossSum) = data.sample(false, miniBatchFraction, 42 + i)  
            .treeAggregate((BDV.zeros[Double](n), 0.0))(  
              seqOp = (c, v) => (c, v) match { case ((grad, loss), (label, features)) =>  
                val l = gradient.compute(features, label, bcWeights.value, Vectors.fromBreeze(grad))  
                (grad, loss + l)  
              },  
              combOp = (c1, c2) => (c1, c2) match { case ((grad1, loss1), (grad2, loss2)) =>  
                (grad1 += grad2, loss1 + loss2)  
              })  

          /** 
           * NOTE(Xinghao): lossSum is computed using the weights from the previous iteration 
           * and regVal is the regularization value computed in the previous iteration as well. 
           */  
          stochasticLossHistory.append(lossSum / miniBatchSize + regVal)  
          val update = updater.compute(  
            weights, Vectors.fromBreeze(gradientSum / miniBatchSize), stepSize, i, regParam)  
          weights = update._1  
          regVal = update._2  
        }  

每次根據MiniBatchFraction指定的比例隨機采樣相應數量的樣本，然后調用Gradient.Compute逐個計算Gradient和 Loss并累加在一起，得到這一輪迭代總的Gradient和Loss的變化。然后調用Updater.Compute更新Weights矩陣和 RegVal（正則化項）。也就是說Gradient.Compute只是利用當前Weights向量計算Gradient和Loss的變化，而 Updater.Compute則根據這個變化以及正則化項計算更新之后的Weights和RegVal。

SGD算法支持使用L1和L2正則化方法。

注意到這里Weights向量在下一輪計算的過程中每個參與計算的Executor都需要，所以使用了Broadcast變量把它分發到每個節點，提高了計算效率。

LBFGS：

LBFGS 優化方法的核心實現在LBFGS.runLBFGS函數里面。LBFGS的實現比SGD更加依賴Breeze庫，它的迭代框架都是使用的Breeze的 LBFGS的實現，只是實現了自己的名為CostFun的DiffFunction。大家可以去LBGFS.CostFun函數中看看Loss Function和Gradient的計算方法與SGD算法如出一轍，也是利用了RDD的Aggregate操作。

和SGD不同，目前 LBFGS只支持L2正則化，不支持L1正則化。 其實在Breeze庫里面有LBFGS + L1正則化的實現OWLQN（OWLQN算法默認自帶L1正則化，所以在傳入的參數DiffFunction中不需要顯示定義正則化項，只需要定義 Loss Function即可）是可以把它引入MLlib里面完成LBFGS+L1的功能。 這個在社區也有討論，DB Tsai等人正在做這方面的工作。等不及的同學也可以嘗試下我們自己修改的版本，引入了LBFGS+L1的功能的代碼。

SGD和LBFGS兩種算法的比較：

網上的資料都告訴我們說LBFGS比SGD更容易收斂，效果更好，大家可以親自嘗試下。例如選擇Logistic Regression算法，選取同一個數據集，在做Training和Test集合的分配的時候也要一致。然后把生成的Training和Test的 RDD分別丟到LogisticRegressionWithSGD和LogisticRegressionWithLBFGS兩種具體實現算法里。其他參數要一致（例如都選擇L2正則化，regParam=1.0）然后比較效果。

怎么比較效果的好壞呢？分類問題就是那些指標 Precision/Recall/F1-score/Area Under ROC等。這里面有個需要注意的問題，MLlib的實現里面，SGD優化的終止條件是通過指定NumIterations也就是迭代次數終止的；而 LBFGS優化的終止條件是通過指定ConvergenceTol（兩次迭代Loss Function變化的容忍度）和MaxnumIterations（最大迭代次數）來終止的。

為了達到比較這兩種優化方法的目的，需要定義一個統一的指標。由于我們優化的目標是讓Loss Function+正則化項 最小，所以這個就是統一的指標。MLlib的日志里面會打印出迭代的最后10次的Loss大小：

//GradientDescent

    logInfo("GradientDescent.runMiniBatchSGD finished. Last 10 stochastic losses %s".format(  

    stochasticLossHistory.takeRight(10).mkString(", ")))  

    logInfo("LBFGS.runLBFGS finished. Last 10 losses %s".format(  

    lossHistory.takeRight(10).mkString(", ")))  

我在一個數據集上測試，其他參數都一致（L2正則化，RegParam=1.0，MiniBatchFraction=1.0）的情況下，LBFGS需要 14次就可以達到收斂；而對于SGD（實際上MiniBatchFraction=1.0的已經不是純種SGD了）算法大約需要10000次循環才能達到相似的Loss大小。而且最后兩種算法得到的Weights（權重矩陣）是一樣的。

3，如何解讀和分析訓練出來的模型

訓練出來的模型實際上就是一個Weights向量，現在MLlib的GeneralizedLinearModel的成員變量Weights和 Intercept都是Public的了。 在訓練好GeneralizedLinearModel之后，可以直接把Weights和Intercept打印出來，或者進行一些計算找出權重最大的幾個維度。

這里有個問題，如果我的Loss Function和Gradient是確定的，那么使用不同的優化方法求出的Weights是不是應該是一樣的？例如對同一個數據集，使用 LogisticRegressionWithSGD和LogisticRegressionWithLBFGS分別訓練出的模型是否應該是一致的？怎么衡量效果好壞？

對于這個問題，我想首先要看幾點：

1)  確認兩種方法使用的正則化和其他參數是一致的。

2)  通過日志查看兩種優化方法的Loss Function+正則化項最后是否收斂?是否收斂到接近的值？

3)  如果有某種優化算法的Loss不收斂，說明那種方法的迭代次數不夠（SGD）或者ConvergenceTol設置的太大（LBFGS），調整參數重試。如果收斂到相差較大的值，十有八九你的算法有問題。

4)  如果兩種優化方法的Loss Function+正則化項收斂且到接近的值，那么就要看看傳統指標Precision/Recall/Area Under ROC。如果Area Under ROC接近1.0，說明你的數據完全線性可分或者過擬合了，這個時候打印出兩種方法得到的Weights應該是類似倍數關系的；如果Area Under ROC不是接近1.0，那么說明你的數據是真實的數據，這個時候兩種方法得到的Weights應該是一致的。

5)  接下來就可以使用上述的思路去調節參數優化Precision/Recall等指標了。

4，預處理

其實上面的工作還只是停留在學習算法的階段，拿過來一些公開的Dataset或者Benchmark來跑跑，看看效果。這些數據集往往都是經過了別人的一些加工和處理，在實際工作中這一部分也是需要我們來做的，這就是數據預處理。數據預處理特別繁瑣，但是對機器學習模型效果的好壞卻非常重要。

預處理主要包括Normalization，Scale，Outlier-Detection，正負樣本均衡等。例如遇到一個數據正負樣本比例9:1，這樣的數據直接丟到模型里面顯然會讓模型更偏向正例做預測。解決這個問題的方法挺多的，最簡單的例子就是正例采樣、負例冗余，使兩者達到接近平衡。

MLlib 現階段主要精力還是在核心算法上，對預處理這部分做的不是很好，這也提高了使用門檻。在1.1.0版本開始增加了一個新的Package叫 Feature，里面大多是Preprocssing函數，包括Normalizer和StandardScaler等。例如 StandardScaler能夠把Feature按照列轉換成Mean=0，Standard deviation=1的正態分布。

數據輸入支持普通文本文件和LIBSVM格式。在1.1.0版本之前，輸入的Label可以是+1/-1會被自動映射成1.0/0.0，但是從1.1.0開始貌似只支持輸入1.0/0.0的Label了，這個和我們以前常見的LIBSVM格式的數據集不一樣，需要注意。這個改動應該主要考慮的是對以后多分類的支持吧。

5，MLlib中的Vector和線性代數運算

不知道大家有沒有注意到一個問題，就是MLlib底層的矩陣運算使用了Breeze庫，Breeze庫提供了Vector/Matrix的實現以及相應計算的接口（Linalg）。但是在MLlib里面同時也提供了Vector和Linalg等的實現（目前只是對Breeze做了一層包裝）。在所有的 MLlib的函數里面的參數傳遞都是使用Mllib自己的Vector，而且在函數內的矩陣計算又通過ToBreeze.ToDenseVector變成 Breeze的形式進行運算。這樣做的目的一是保持自己函數接口的穩定性，不會因為Breeze的變化而變化；另外一個就是可以把Distributed Matrix作為一種Matrix的實現而被使用。

6，開發環境

Spark集群（Standalone、Yarn-Client、Yarn-Cluster、單機調試環境）。

我主要使用Scala開發，IDE為Intellij IDEA，安裝Scala插件。

開發一個Project可以使用Maven或者SBT編譯，都可以通過IDEA創建相應的工程。 Maven編譯的話和Java的Maven工程沒啥區別，主要是修改Pom.xml文件；使用SBT編譯的話，主要是修改Build.sbt文件。

Build.sbt的格式網上有很多資料了，簡單說下需要注意的問題：

1) 必須每隔一行寫新的內容；

2)LibraryDependencies 后面%%和%的區別：ArtifactId后面帶/不帶版本號；

3)LibraryDependencies 后面可以使用”Provided”使其在Assembly打包的時候不被打入包中。

開發一個基于Spark和MLlib的機器學習Job，主要依賴的兩個LibraryDependencies就是Spark-Core和Spark-Mllib。

其實使用Scala開發Spark程序最重要的一點就是要知道你寫的代碼中哪些是RDD的操作，哪些是在RDD內部的操作，哪些是 Transform，哪些是Actions，哪個地方會形成一個Stage。這些搞清楚之后就明白了哪些Code是在Driver上執行的，哪些是在 Executor上并行執行的。另外就是哪些資源相關的參數，像Executor-Memory和Num-Executors等。關于這方面的內容在后面的介紹Tree和Random Forest的博客中討論。

原文鏈接：mllib實踐經驗    （責編/劉亞瓊）