從Spark MLlib到美圖機器學習框架實踐

摘要： MLlib 是 Apache Spark 的可擴充套件機器學習庫，旨在簡化機器學習的工程實踐工作，並方便擴充套件到更大規模的資料集。  /機器學習簡介 /  在深入介紹 Spark MLlib 之前先了解機器學習，根據維基百科的介紹，機器學習有下面幾種定義： ...

MLlib 是 Apache Spark 的可擴充套件機器學習庫，旨在簡化機器學習的工程實踐工作，並方便擴充套件到更大規模的資料集。

 /機器學習簡介 / 

在深入介紹 Spark MLlib 之前先了解機器學習，根據維基百科的介紹，機器學習有下面幾種定義：

• 機器學習是一門人工智慧的科學，該領域的主要研究物件是人工智慧，特別是如何在 經驗 學習中改善具體 演算法 的 效能 ；

• 機器學習是對能通過 經驗 自動改進的計算機 演算法 的研究；

• 機器學習是用資料或以往的 經驗 ，以此優化計算機程式的 效能 標準；

• 一種經常引用的英文定義是「A computer program is said to learn from experience  E with respect to some class of tasks T and performance measure P, if its performance at tasks in T, as measured by P, improves with experience E.」。

*加粗的是重點/加粗的是重點/加粗的是重點

其實在「美圖資料技術團隊」之前的科普文章 ofollow,noindex">貝葉斯概率模型一覽 曾介紹過，機器學習狹義上是指代統計機器學習，統計學習根據任務型別可以分為監督學習、半監督學習、無監督學習、增強學習等。

機器學習常用的演算法可以分為以下種類：

1.構造間隔理論分佈：人工神經網路、決策樹、感知器、支援向量機、整合學習AdaBoost、降維與度量學習、聚類、貝葉斯分類器；

 / Spark MLlib / 

在上文我們曾提到機器學習的重點之一是「經驗」，而對於計算機而言經驗往往需要經過多輪迭代計算才能得到，而 Spark 擅長迭代計算，正好符合機器學習這一特性。在 Spark 官網上展示了邏輯迴歸演算法在 Spark 和 Hadoop 上執行效能比較，從下圖可以看出 MLlib 比 MapReduce 快了 100 倍。

Spark MLlib 主要包括以下幾方面的內容：

• 學習演算法：分類、迴歸、聚類和協同過濾；

• 特徵處理：特徵提取、變換、降維和選擇；

• 管道(Pipeline)：用於構建、評估和調整機器學習管道的工具；

• 永續性：儲存和載入演算法，模型和管道；

• 實用工具：線性代數，統計，最優化，調參等工具。

上表總結了 Spark MLlib 支援的功能結構，可以看出它所提供的演算法豐富，但演算法種類較少並且老舊，因此 Spark MLlib 在演算法上支援與 kylin 專案有些脫節，它的主要功能更多是與特徵相關的。

ML Pipelines

從 Spark 2.0 開始基於 RDD 的 API 進入維護模式，Spark 的主要機器學習API 現在是基於 DataFrame 的 API spark.ml，借鑑 Scikit-Learn 的設計提供了 Pipeline 套件，以構建機器學習工作流。 ML Pipelines 提供了一套基於 DataFrame 構建的統一的高階 API ，可幫助使用者建立和調整實用的機器學習流程。

*「Spark ML」不是官方名稱，偶爾用於指代基於 MLlib DataFrame 的 API

首先了解 ML Pipelines 內幾個重要元件。

DataFrame

DataFrame 讓 Spark 具備了處理大規模結構化資料的能力。

RDD 是分散式  Java 物件的集合，物件的內部資料結構對於 RDD 而言不可知。DataFrame 是一種以 RDD 為基礎的分散式資料集，RDD 中儲存了 Row 物件，Row 物件提供了詳細的結構資訊，即模式（schema），使得 DataFrame 具備了結構化資料的能力。

Transforme

Transformer 通常是一個數據/特徵變換的類，或一個訓練好的模型。

每個 Transformer 都有 transform 函式，用於將一個 DataFrame 轉換為另一個 DataFrame 。一般 transform 的過程是在輸入的 DataFrame 上新增一列或者多列 ，Transformer.transform也是惰性執行，只會生成新的 DataFrame 變數，而不會去提交 job 計算 DataFrame 中的內容。

Estimator

Estimator 抽象了從輸入資料學習模型的過程，每個 Estimator 都實現了 fit 方法，用於給定 DataFrame 和 Params 後，生成一個 Transformer（即訓練好的模型），每當呼叫 Estimator.fit() 後，都會產生 job 去訓練模型，得到模型引數。

Param

可以通過設定 Transformer 或 Estimator 例項的引數來設定模型引數，也可以通過傳入 ParamMap 物件來設定模型引數。

Pipeline

Pipeline 定義了一組資料處理流程，可以在 Pipeline 中加入 Transformer、Estimator 或另一個 Pipeline。Pipeline 繼承自 Estimator，呼叫 Pipeline.fit 方法後返回一個 Transformer——PipelineModel；PipelineModel 繼承自 Transformer，用於將輸入經過 Pipeline 的各個 Transformer 的變換後，得到最終輸出。

Spark MLlib 典型流程如下：

• 構造訓練資料集

• 構建各個 Stage

• Stage 組成 Pipeline

• 啟動模型訓練

• 評估模型效果

• 計算預測結果

通過一個 Pipeline 的文字分類示例來加深理解：

import org.apache.spark.ml.{Pipeline, PipelineModel}
import org.apache.spark.ml.classification.LogisticRegression
import org.apache.spark.ml.feature.{HashingTF, Tokenizer}
import org.apache.spark.ml.linalg.Vector
import org.apache.spark.sql.Row

// Prepare training documents from a list of (id, text, label) tuples.
val training = spark.createDataFrame(Seq(
 (0L, "a b c d e spark", 1.0),
 (1L, "b d", 0.0),
 (2L, "spark f g h", 1.0),
 (3L, "hadoop mapreduce", 0.0)
)).toDF("id", "text", "label")

// Configure an ML pipeline, which consists of three stages: tokenizer, hashingTF, and lr.
val tokenizer = new Tokenizer()
 .setInputCol("text")
 .setOutputCol("words")
val hashingTF = new HashingTF()
 .setNumFeatures(1000)
 .setInputCol(tokenizer.getOutputCol)
 .setOutputCol("features")
val lr = new LogisticRegression()
 .setMaxIter(10)
 .setRegParam(0.001)
val pipeline = new Pipeline()
 .setStages(Array(tokenizer, hashingTF, lr))

// Fit the pipeline to training documents.
val model = pipeline.fit(training)

// Now we can optionally save the fitted pipeline to disk
model.write.overwrite().save("/tmp/spark-logistic-regression-model")

// We can also save this unfit pipeline to disk
pipeline.write.overwrite().save("/tmp/unfit-lr-model")

// And load it back in during production
val sameModel = PipelineModel.load("/tmp/spark-logistic-regression-model")

// Prepare test documents, which are unlabeled (id, text) tuples.
val test = spark.createDataFrame(Seq(
 (4L, "spark i j k"),
 (5L, "l m n"),
 (6L, "spark hadoop spark"),
 (7L, "apache hadoop")
)).toDF("id", "text")

// Make predictions on test documents.
model.transform(test)
 .select("id", "text", "probability", "prediction")
 .collect()
 .foreach { case Row(id: Long, text: String, prob: Vector, prediction: Double) =>
   println(s"($id, $text) --> prob=$prob, prediction=$prediction")
 }

模型選擇與調參

Spark MLlib 提供了 CrossValidator 和 TrainValidationSplit 兩個模型選擇和調參工具。模型選擇與調參的三個基本元件分別是 Estimator、ParamGrid 和 Evaluator，其中 Estimator 包括演算法或者 Pipeline；ParamGrid 即 ParamMap 集合，提供引數搜尋空間；Evaluator 即評價指標。

CrossValidator

via https://github.com/JerryLead/blogs/blob/master/BigDataSystems/Spark/ML/Introduction%20to%20MLlib%20Pipeline.md

CrossValidator 將資料集按照交叉驗證數切分成 n 份，每次用 n-1 份作為訓練集，剩餘的作為測試集，訓練並評估模型，重複 n 次，得到 n 個評估結果，求 n 次的平均值作為這次交叉驗證的結果。接著對每個候選 ParamMap 重複上面的過程，選擇最優的 ParamMap 並重新訓練模型，得到最優引數的模型輸出。

:chestnut:舉個例子：

// We use a ParamGridBuilder to construct a grid ofparameters to search over. // With 3 values for hashingTF.numFeatures and 2 values for lr.regParam, // this grid will have 3 x 2 = 6 parametersettings for CrossValidator to choose from. val paramGrid = new ParamGridBuilder()  .addGrid(hashingTF.numFeatures, Array(10, 100, 1000))  .addGrid(lr.regParam, Array(0.1, 0.01))  .build() // We now treat the Pipeline as an Estimator, wrapping it in a CrossValidator instance. // This will allow us to jointly choose parameters for all Pipeline stages. // A CrossValidator requires an Estimator, a set of Estimator ParamMaps, and an Evaluator. // Note that the evaluator here is a BinaryClassificationEvaluator and its default metric // is areaUnderROC. val cv = new CrossValidator()  .setEstimator(pipeline)  .setEvaluator(new BinaryClassificationEvaluator)  .setEstimatorParamMaps(paramGrid)  .setNumFolds(2)  // Use 3+ in practice  .setParallelism(2)  // Evaluate up to 2 parametersettings in parallel // Run cross-validation, and choose the best set of parameters. val cvModel = cv.fit(training) // Prepare test documents, which are unlabeled (id, text) tuples. val test = spark.createDataFrame(Seq(  (4L, "spark i j k"),  (5L, "l m n"),  (6L, "mapreduce spark"),  (7L, "apache hadoop") )).toDF("id", "text") // Make predictions on test documents. cvModel uses the best model found (lrModel). cvModel.transform(test)  .select("id", "text", "probability", "prediction")  .collect()  .foreach { case Row(id: Long, text: String, prob: Vector, prediction: Double) =>    println(s"($id, $text) --> prob=$prob, prediction=$prediction")  }

TrainValidationSplit

TrainValidationSplit 使用 trainRatio引數將訓練集按照比例切分成訓練和驗證集，其中 trainRatio 比例的樣本用於訓練，剩餘樣本用於驗證。

與 CrossValidator 不同的是，TrainValidationSplit 只有一次驗證過程，可以簡單看成是 CrossValidator 的 n 為 2 時的特殊版本。

:chestnut:舉個例子：

import org.apache.spark.ml.evaluation.RegressionEvaluator import org.apache.spark.ml.regression.LinearRegression import org.apache.spark.ml.tuning.{ParamGridBuilder, TrainValidationSplit} // Prepare training and test data. val data = spark.read.format("libsvm").load("data/mllib/sample_linear_regression_data.txt") val Array(training, test) = data.randomSplit(Array(0.9, 0.1), seed = 12345) val lr = new LinearRegression()    .setMaxIter(10) // We use a ParamGridBuilder to construct a grid ofparameters to search over. // TrainValidationSplit will try all combinations of values and determine best model using // the evaluator. val paramGrid = new ParamGridBuilder()  .addGrid(lr.regParam, Array(0.1, 0.01))  .addGrid(lr.fitIntercept)  .addGrid(lr.elasticNetParam, Array(0.0, 0.5, 1.0))  .build() // In this case the estimator is simply the linear regression. // A TrainValidationSplit requires an Estimator, a set of Estimator ParamMaps, and an Evaluator. val trainValidationSplit = new TrainValidationSplit()  .setEstimator(lr)  .setEvaluator(new RegressionEvaluator)  .setEstimatorParamMaps(paramGrid)  // 80% of the data will be used for training and the remaining 20% for validation.  .setTrainRatio(0.8)  // Evaluate up to 2parametersettings in parallel  .setParallelism(2) // Run train validation split, and choose the best set ofparameters. val model = trainValidationSplit.fit(training) // Make predictions on test data. model is the model with combination ofparameters // that performed best. model.transform(test)  .select("features", "label", "prediction")  .show()

實現自定義 Transformer

繼承自 Transformer 類，實現 transform 方法，通常是在輸入的 DataFrame 上新增一列或多列。

對於單輸入列，單輸出列的 Transformer 可以繼承自 UnaryTransformer 類，並實現其中的 createTransformFunc 方法，實現對輸入列每一行的處理，並返回相應的輸出。

 / 自研機器學習框架 / 

機器學習技術日新月異，卻缺少高效靈活的框架降低新技術的調研成本，而經驗與技術往往需要通過框架和工具來沉澱，並且演算法人員常常受限於算力，導致離線證明有效的模型，因為預估時間複雜度過高而無法上線。

據此美圖資料技術團隊以「開發簡單靈活的機器學習工作流，降低演算法人員的新演算法調研成本及工程人員的維護成本，並且提供常用的領域內解決方案，將經驗沉澱」的目標搭建了一套量身定製的機器學習框架用以解決上述問題，尤其是解決在推薦演算法相關任務上遇到的問題。該框架總共包括 3 個元件：Spark Feature、Bamboo 與 Online Scorer。

Spark Feature：訓練樣本生產

該元件主要用於訓練樣本的生產，實現了靈活高效的樣本特徵編碼，可以實現將任意特徵集合放在同一個空間進行編碼，不同特徵集合共享編碼空間；為此我們提出了兩個概念：第一個是「域」，用於定義共享相同建模過程的一組特徵；第二個是「空間」，用於定義共享相同編碼空間的一組域。

上圖示例中的「Old」展示了在沒有“域”和“空間”概念下的樣本特徵編碼，所有特徵從 1 開始編號；「New」展示了將 age 和 gender 分別放到 age 域和 gender 域後，兩個域分別從 1 開始編碼，互不影響。

Spark Feature 最終採用 TFRecords 作為訓練樣本的儲存格式。

Bamboo：模型定義與訓練

該元件主要為了實現可擴充套件、高效、簡單快速的模型定義與訓練。為此，在設計 Bamboo 時我們遵循以下原則：

1.layer 之間通過 tensor 進行互動，layer 的輸入是 tensor，輸出也是 tensor；

2.為了最大限度地提高離線與線上效率，沒有采用太多高階 api，如 keras，大多數模型與元件基於Tensorflow 底層 api 開發，並且根據Tensorflow 官方的效能優化指南對程式碼進行優化；

3.提供 online-offline 的建模框架，複雜計算放到離線，線上只進行輕量計算，使得複雜模型更易上線；

4.封裝資料載入、模型訓練與匯出、效果評估以及提供了各種輔助工具，使用者只需要定義前向推理網路，同時封裝了大量的常用 layer，模型定義更快捷。

Online Scorer：線上預測服務

Online Scorer的目標是提供一個統一，高效的線上推理服務，可以同時支援tensorflow，pytorch，xgboost等各種主流建模框架匯出的模型。目前這塊工作還在進行中，具體實現方案細節，我們放到後面的專題文章介紹。

以上就是美圖自研機器學習框架的簡要介紹，歡迎持續關注「美圖資料技術團隊」，後續將帶來該平臺的詳細介紹。