大資料開發之機器學習總結（Spark Mllib）（四）

背景

1. 在大資料和機器學習交叉的領域，如果公司選擇了hadoop生态，結合spark架構，則spark 的mllib用于機器學習實際應用就是不二選擇了。
2. 團隊有spark基礎，學習和适用門檻低。但如果選擇python生态，則需要團隊有python基礎，另外個人認為，python工程化對比java生态還是差了那麼一些意思。

1. Spark MLLib簡介

1. spark的mllib目前支援4種常見機器學習問題，分類，回歸，聚類，協同過濾。

2. 

3. mllib本身還是基于RDD算子實作，是以天生就可以和spark sql，spark streaming無縫結成，也就是spark的四個子產品spark sql，spark streaming， spark mllib， graphX可以直接解決大資料資料計算的90%以上場景，覆寫結構化資料處理，流式資料處理，機器學習，圖計算。
4. mllib的架構圖

   

3個組成部分

底層基礎：包括Spark的運作庫、矩陣庫和向量庫；

算法庫：包含廣義線性模型、推薦系統、聚類、決策樹和評估的算法；

實用程式：包括測試資料的生成、外部資料的讀入等功能。

1. mllib底層基礎

• 基礎部分主要包括向量接口Vector和矩陣接口Matrix，這兩種接口都會使用Scala語言基于Netlib和BLAS/LAPACK開發的線性代數庫Breeze。

這裡結合另外一篇數學基礎部落格，可以直到如果要處理分類，聚類，協同過濾，很多時候就是将事物特征向量提取出來，然後适用算法對向量做相似度或者空間距離計算

而矩陣可以看成是向量集合，更友善向量的計算，因為是一批向量組合為一個矩陣。

• MLlib支援本地的密集向量和稀疏向量，并且支援标量向量。

  注意，在海量資料處理時，如果适用類似數組等集合表示向量，但如果一個10000長度的數組中大部分資料都是相同的如0，少量是1，則沒必要真的建立10000長度數組，直接以坐标數組和對應坐标值數組表示會更加節省空間。

  如Array(1,8,9,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,18,0,0,0,0,10,0,0,0,0,0,0,0,0,0,11,0,0,0,0,0,89,0,0,0,0,0);這時候就可以采取比這種更加節省記憶體方式，如tuple: ( 1000, Array[0,1,2,54,63,86,90], Array[1,8,9,18,10,11,89] )，就可以表示一個1000長度數組，index對應是Array[0,1,2,54,63,86,90]，對應值是Array[1,8,9,18,10,11,89] 。其他地方都是0，這種叫做稀疏性向量

  類似[1,8,9,76,1,23,78,66,12,35,0,0,0,65,87,8,276,28,8,88,98]，這種稱之為密集型向量

spark中，稀疏和密集型向量對象就是：DenseVector 和SparseVector

示例

稀疏和密集向量在計算和存儲時資料對比

疏矩陣在含有大量非零元素的向量Vector計算中會節省大量的空間并大幅度提高計算速度，如下圖所示。

标量LabledPoint在實際中也被大量使用，例如判斷郵件是否為垃圾郵件時就可以使用類似于以下的代碼：

可以把表示為1.0的判斷為正常郵件，而表示為0.0則作為垃圾郵件來看待。

對于矩陣Matrix而言，本地模式的矩陣如下所示。

• MLlib同時支援本地矩陣和分布式矩陣，支援的分布式矩陣分為RowMatrix、IndexedRowMatrix、CoordinateMatrix等。

在大資料處理時，分布式矩陣的引入更有意義，當然也支援本地矩陣。

分布式計算最主要就是資料切分和任務切分，這樣可以充分利用各個節點的并行計算能力。其他考慮點則是優化資料切分政策，資料傳輸政策，中間計算結果儲存，計算結果儲存，任務切分，任務排程等等因素。

1. 矩陣程式設計接口

• RowMatrix直接通過RDD[Vector]來定義并可以用來統計平均數、方差、協同方差等：

  

  

平均數，如何計算就不贅述

均值描述的是樣本集合的中間點，它告訴我們的資訊是很有限的

标準差

标準差給我們描述的則是樣本集合的各個樣本點到均值的距離之平均

方差

方差則僅僅是标準差的平方

協方差

• IndexedRowMatrix是帶有索引的Matrix，但其可以通過toRowMatrix方法來轉換為RowMatrix，進而利用其統計功能，代碼示例如下所示。

  

• CoordinateMatrix常用于稀疏性比較高的矩陣，是由RDD[MatrixEntry]來建構的，MatrixEntry是一個Tuple類型的元素，其中包含行、列和元素值，代碼示例如下所示：

  

1. 特征向量處理（規範化）

   把原始資料中的值直接轉入向量，各特征的值量綱可能差距很大，對算法的效果産生巨大負面效應（某些特征可能會掩蓋掉其他特征的作用）是以需要對向量進行縮放（規範化、歸一化）,mllib中有四種工具

• 範數規範器：Normalizer

  針對一行來操作！

  .setP(2) 2階P範數

  

import org.apache.spark.ml.feature.Normalizer

// 正則化每個向量到1階範數
val normalizer = new Normalizer()
  .setInputCol("features")
  .setOutputCol("normFeatures")
  .setP(1.0)

val l1NormData = normalizer.transform(dataFrame)
println("Normalized using L^1 norm")
l1NormData.show()

// 将每一行的規整為1階範數為1的向量，1階範數即所有值絕對值之和。
+-----+---------------------+-----------------------------------+
| id  |   features   |      normFeatures   |
+-----+----------------------+----------------------------------+
|  0| [1.0,0.5,-1.0]  |    [0.4,0.2,-0.4]       |
|  1| [2.0,1.0,1.0]   |   [0.5,0.25,0.25]      |
|  2| [4.0,10.0,2.0]  |[0.25,0.625,0.125]      |
+---+------------------------+---------------------------------+

// 正則化每個向量到無窮階範數
val lInfNormData = normalizer.transform(dataFrame, normalizer.p -> Double.PositiveInfinity)
println("Normalized using L^inf norm")
lInfNormData.show()

// 向量的無窮階範數即向量中所有值中的最大值
+---+-------------------+----------------------+
| id|  features   |  normFeatures|
+---+------------------+-----------------------+
|  0|[1.0,0.5,-1.0]|[1.0,0.5,-1.0]   |
|  1| [2.0,1.0,1.0]| [1.0,0.5,0.5]   |
|  2|[4.0,10.0,2.0]| [0.4,1.0,0.2]   |
+---+-------------------+----------------------+
           

• 标準差規範器：StandardNormalizer

  将特征标準化為機關标準差或是0均值，或是0均值機關标準差。

  将每一列的标準差限制在0-1之間，進而倒推各特征值的縮放

  标準差公式：

  

import org.apache.spark.ml.feature.StandardScaler

val scaler = new StandardScaler()
  .setInputCol("features")
  .setOutputCol("scaledFeatures")
  .setWithStd(true)
  .setWithMean(false)

// Compute summary statistics by fitting the StandardScaler.
val scalerModel = scaler.fit(dataFrame)

// Normalize each feature to have unit standard deviation.
val scaledData = scalerModel.transform(dataFrame)
scaledData.show

// 将每一列的标準差縮放到1。
+---+--------------+------------------------------------------------------------+
|id |features      |scaledFeatures                                              |
+---+--------------+------------------------------------------------------------+
|0  |[1.0,0.5,-1.0]|[0.6546536707079772,0.09352195295828244,-0.6546536707079771]|
|1  |[2.0,1.0,1.0] |[1.3093073414159544,0.1870439059165649,0.6546536707079771]  |
|2  |[4.0,10.0,2.0]|[2.618614682831909,1.870439059165649,1.3093073414159542]    |
+---+--------------+------------------------------------------------------------+
           

• 值域範圍縮放器：MinMaxScaler

  将一列特征的（| 最大值-最小值 |）作為分母，（特征x-最小值）作為分子

import org.apache.spark.ml.feature.MinMaxScaler

val scaler = new MinMaxScaler()
  .setInputCol("features")
  .setOutputCol("scaledFeatures")

// Compute summary statistics and generate MinMaxScalerModel
val scalerModel = scaler.fit(dataFrame)

// rescale each feature to range [min, max].
val scaledData = scalerModel.transform(dataFrame)
println(s"Features scaled to range: [${scaler.getMin}, ${scaler.getMax}]")
scaledData.select("features", "scaledFeatures").show

// 每維特征線性地映射，最小值映射到0，最大值映射到1。
+--------------+-----------------------------------------------------------+
|features      |scaledFeatures                                             |
+--------------+-----------------------------------------------------------+
|[1.0,0.5,-1.0]|[0.0,0.0,0.0]                                              |
|[2.0,1.0,1.0] |[0.3333333333333333,0.05263157894736842,0.6666666666666666]|
|[4.0,10.0,2.0]|[1.0,1.0,1.0]                                              |
+--------------+-----------------------------------------------------------+
           

• 最大絕對值縮放器：MaxAbsScaler

  将一列特征的最大絕對值作為分母，（特征x）作為分子

import org.apache.spark.ml.feature.MaxAbsScaler

val scaler = new MaxAbsScaler()
  .setInputCol("features")
  .setOutputCol("scaledFeatures")

// Compute summary statistics and generate MaxAbsScalerModel
val scalerModel = scaler.fit(dataFrame)

// rescale each feature to range [-1, 1]
val scaledData = scalerModel.transform(dataFrame)
scaledData.select("features", "scaledFeatures").show()

// 每一維的絕對值的最大值為[4, 10, 2]
+--------------+----------------+                                               
|      features|  scaledFeatures|
+--------------+----------------+
|[1.0,0.5,-1.0]|[0.25,0.05,-0.5]|
| [2.0,1.0,1.0]|   [0.5,0.1,0.5]|
|[4.0,10.0,2.0]|   [1.0,1.0,1.0]|
+--------------+----------------+
           

上述介紹了MLlib的底層架構，向量和矩陣程式設計接口，還有特征向量規範化處理

2. 模型評估

在機器學習中，如何評估一個訓練出來的模型是否好，有多好是需要盡可能資料來量化的，這時候就需要使用算法來評估以及量化模型的準确程度

2.1 混淆矩陣

1. 矩陣，可以了解為就是一張表格，混淆矩陣其實就是一張表格而已。

以分類模型中最簡單的二分類為例，對于這種問題，我們的模型最終需要判斷樣本的結果是0還是1，或者說是positive還是negative。

我們通過樣本的采集，能夠直接知道真實情況下，哪些資料結果是positive，哪些結果是negative。同時，我們通過用樣本資料跑出分類型模型的結果，也可以知道模型認為這些資料哪些是positive，哪些是negative。

我們就能得到這樣四個基礎名額，我稱他們是一級名額（最底層的）：

真實值是positive，模型認為是positive的數量（True Positive=TP）

真實值是positive，模型認為是negative的數量（False Negative=FN）：這就是統計學上的第一類錯誤（Type I Error）

真實值是negative，模型認為是positive的數量（False Positive=FP）：這就是統計學上的第二類錯誤（Type II Error）

真實值是negative，模型認為是negative的數量（True Negative=TN）

1. 預測性分類模型，肯定是希望越準越好。那麼，對應到混淆矩陣中，那肯定是希望TP與TN的數量大，而FP與FN的數量小。是以當我們得到了模型的混淆矩陣後，就需要去看有多少觀測值在第二、四象限對應的位置，這裡的數值越多越好；反之，在第一、三象限對應位置出現的觀測值肯定是越少越好。
2. 混淆矩陣裡面統計的是個數，有時候面對大量的資料，光憑算個數，很難衡量模型的優劣。是以混淆矩陣在基本的統計結果上又延伸了如下4個名額，我稱他們是二級名額（通過最底層名額加減乘除得到的）：

   

   準确率（Accuracy）—— 針對整個模型

   精确率（Precision）

   靈敏度（Sensitivity）：就是召回率（Recall）

   特異度（Specificity）

// 模型評估 混淆矩陣
val rdd = prediction.rdd.map(row=>{
  val label = row.getAs[Double]("label")
  val prediction = row.getAs[Double]("prediction")
  (prediction,label)
})
val matrix: Matrix = new MulticlassMetrics(rdd).confusionMatrix
           

2.2 AUC曲線(分類算法)

1. ROC曲線，即為一條ROC曲線（該曲線的原始資料第三部分會介紹）。現在關心的是：

   橫軸：False Positive Rate（假陽率，FPR）

   縱軸：True Positive Rate（真陽率，TPR）

   ROC 就是假陽率和真陽率之間的關系曲線

   

假陽率，簡單通俗來了解就是預測為正樣本但是預測錯了的可能性，顯然，我們不希望該名額太高。

真陽率，則是代表預測為正樣本但是預測對了的可能性，當然，我們希望真陽率越高越好。

顯然，ROC曲線的橫縱坐标都在[0,1]之間，自然ROC曲線的面積不大于1。現在我們來分析幾個特殊情況，進而更好地掌握ROC曲線的性質：

(0,0)：假陽率和真陽率都為0，即分類器全部預測成負樣本

(0,1)：假陽率為0，真陽率為1，全部完美預測正确，HAPPY

(1,0)：假陽率為1，真陽率為0，全部完美預測錯誤，悲劇

(1,1)：假陽率和真陽率都為1，即分類器全部預測成正樣本

TPR＝FPR，斜對角線，預測為正樣本的結果一半是對的，一半是錯的，代表随機分類器的預測效果

于是，我們可以得到基本的結論：ROC曲線在斜對角線以下，則表示該分類器效果差于随機分類器，反之，效果好于随機分類器，當然，我們希望ROC曲線盡量除于斜對角線以上，也就是向左上角（0,1）凸。

1. AUC實際上就是ROC曲線下的面積

   ROC曲線一定程度上可以反映分類器的分類效果，但是不夠直覺，我們希望有這麼一個名額，如果這個名額越大越好，越小越差，于是，就有了AUC

AUC直覺地反映了ROC曲線表達的分類能力。

AUC ＝ 1，代表完美分類器

0.5 < AUC < 1，優于随機分類器

0 < AUC < 0.5，差于随機分類器

// 模型評估
val ev = new BinaryClassificationEvaluator()
    .setLabelCol("label")
  .setMetricName("areaUnderROC")
val roc: Double = ev.evaluate(prediction)
println(roc)
           

2.3 回歸算法評估

注意，回歸算法其實就是找出已有資料中，資料輸入和輸出之間關系，簡單如線性回歸，就是輸入和輸出之間的線性關系。

回歸算法的輸出結果，不可能用準确率精确率等名額來衡量

應該用預測值和真實值之間的誤差情況來衡量，具體來說，通常有如下幾個名額：

1. 回歸分析評估名額

• RMSE（Root Mean Square Error）均方根誤差

  衡量觀測值與真實值之間的偏差。

  常用來作為機器學習模型預測結果衡量的标準

  

  開根号： 為了讓量綱更直覺

  100元,90元 => RMSE =10元

• MSE（Mean Square Error）均方誤差

  MSE是真實值與預測值的內插補點的平方然後求和平均。

  通過平方的形式便于求導，是以常被用作線性回歸的損失函數

  

  100元,90元 => MSE =100
• R2

  

• MAE（Mean Absolute Error）平均絕對誤差

  是絕對誤差的平均值。

  可以更好地反映預測值誤差的實際情況。

  

2.4回歸分析評估名額含義

• SSE

  預測值和真值相差的平方和是SSE,也就是誤差平方和，這肯定是越小越好，相當于一個誤差累計。當然這個SSE越接近于0越好

  

  但如果10000的樣本的情況，建立一個A模型，這個模型的SSE是100，100個樣本的情況下，建立一個B模型，這個模型的SSE是80。但是不能說B模型比A模型好。是以就引入了MSE
• MSE

  

  就是均方誤差，SSE除以樣本量，平均的預測的值和真值差的平方，平均到每一個預測的Y

  

• RMSE 對MSE開方就是RMSE，也就是均方根
• SSR 表示的是預測值和原始值得均值差得平方和

  

• SST 表示得是原始資料和均值的差的平方和

  

• R2 就是R-square，可以經過公式推導得出SST=SSE+SSR

  

  将R2寫開

  

R2的範圍在0-1之間，越接近1，表示越好

一般衡量線性回歸最好的名額應該就是R2，通常表示模型拟合的好壞。

對R2開根号，就是R，也就是相關系數，也是越近1越好

上面真值和預測值之間的誤差都是做差求平方和，如果将平方和換成取絕對值，也就是MAE,RMAE,也就是不是square，變為absolute

val evaluator4 = new RegressionEvaluator()
  .setLabelCol("label")
  .setPredictionCol("prediction")
  .setMetricName("mae")
val d4: Double = evaluator4.evaluate(prediction)
           

3. 總結

1. 本文主要是spark mllib子產品簡介v，包括mllib的底層架構，向量和矩陣程式設計接口，如何使用mllib做特征向量規範化處理
2. 當模型選擇并訓練好之後，如何進行模型的效果評估，直接基于訓練集或者驗證集并無法得出一個量化結果，這時候就需要使用算法進行評估。這些評估算法區分算法類型，回歸類型，分類類型等
3. 回歸分析評估名額較多，含義做了單獨解釋。注意，如果沒有機器學習相關知識儲備，不了解或者無法深刻了解這些名額和算法是很正常，因為我也遇到了同樣問題，需要補充理論知識同時結合代碼做實踐，這樣了解會更加深刻。