MLlib 是 Apache Spark 的可擴(kuò)展機(jī)器學(xué)習(xí)庫，旨在簡化機(jī)器學(xué)習(xí)的工程實(shí)踐工作，并方便擴(kuò)展到更大規(guī)模的數(shù)據(jù)集。

/ 機(jī)器學(xué)習(xí)簡介 /

在深入介紹 Spark MLlib 之前先了解機(jī)器學(xué)習(xí)，根據(jù)維基百科的介紹，機(jī)器學(xué)習(xí)有下面幾種定義：

• 機(jī)器學(xué)習(xí)是一門人工智能的科學(xué)，該領(lǐng)域的主要研究對象是人工智能，特別是如何在經(jīng)驗(yàn)學(xué)習(xí)中改善具體算法的性能;

• 機(jī)器學(xué)習(xí)是對能通過經(jīng)驗(yàn)自動(dòng)改進(jìn)的計(jì)算機(jī)算法的研究;

• 機(jī)器學(xué)習(xí)是用數(shù)據(jù)或以往的經(jīng)驗(yàn)，以此優(yōu)化計(jì)算機(jī)程序的性能標(biāo)準(zhǔn);

• 一種經(jīng)常引用的英文定義是「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.」。

*加粗的是重點(diǎn)/加粗的是重點(diǎn)/加粗的是重點(diǎn)

其實(shí)在「美圖數(shù)據(jù)技術(shù)團(tuán)隊(duì)」之前的科普文章貝葉斯概率模型一覽曾介紹過，機(jī)器學(xué)習(xí)狹義上是指代統(tǒng)計(jì)機(jī)器學(xué)習(xí)，統(tǒng)計(jì)學(xué)習(xí)根據(jù)任務(wù)類型可以分為監(jiān)督學(xué)習(xí)、半監(jiān)督學(xué)習(xí)、無監(jiān)督學(xué)習(xí)、增強(qiáng)學(xué)習(xí)等。

 

 

機(jī)器學(xué)習(xí)常用的算法可以分為以下種類：

1.構(gòu)造間隔理論分布：人工神經(jīng)網(wǎng)絡(luò)、決策樹、感知器、支持向量機(jī)、集成學(xué)習(xí)AdaBoost、降維與度量學(xué)習(xí)、聚類、貝葉斯分類器;

2.構(gòu)造條件概率：高斯過程回歸、線性判別分析、最近鄰居法、徑向基函數(shù)核;

3.通過再生模型構(gòu)造概率密度函數(shù)：最大期望算法、概率圖模型(貝葉斯網(wǎng)和 Markov 隨機(jī)場)、Generative Topographic Mapping;

4.近似推斷技術(shù)：馬爾可夫鏈、蒙特卡羅方法、變分法;

5.最優(yōu)化算法。

/ Spark MLlib /

在上文我們曾提到機(jī)器學(xué)習(xí)的重點(diǎn)之一是「經(jīng)驗(yàn)」，而對于計(jì)算機(jī)而言經(jīng)驗(yàn)往往需要經(jīng)過多輪迭代計(jì)算才能得到，而 Spark 擅長迭代計(jì)算，正好符合機(jī)器學(xué)習(xí)這一特性。在 Spark 官網(wǎng)上展示了邏輯回歸算法在 Spark 和 Hadoop 上運(yùn)行性能比較，從下圖可以看出 MLlib 比 MapReduce 快了 100 倍。

 

 

Spark MLlib 主要包括以下幾方面的內(nèi)容：

• 學(xué)習(xí)算法：分類、回歸、聚類和協(xié)同過濾;

• 特征處理：特征提取、變換、降維和選擇;

• 管道(Pipeline)：用于構(gòu)建、評估和調(diào)整機(jī)器學(xué)習(xí)管道的工具;

• 持久性：保存和加載算法，模型和管道;

• 實(shí)用工具：線性代數(shù)，統(tǒng)計(jì)，最優(yōu)化，調(diào)參等工具。

 

 

上表總結(jié)了 Spark MLlib 支持的功能結(jié)構(gòu)，可以看出它所提供的算法豐富，但算法種類較少并且老舊，因此 Spark MLlib 在算法上支持與 kylin 項(xiàng)目有些脫節(jié)，它的主要功能更多是與特征相關(guān)的。

ML Pipelines

從 Spark 2.0 開始基于 RDD 的 API 進(jìn)入維護(hù)模式，Spark 的主要機(jī)器學(xué)習(xí) API 現(xiàn)在是基于 DataFrame 的 API spark.ml，借鑒 Scikit-Learn 的設(shè)計(jì)提供了 Pipeline 套件，以構(gòu)建機(jī)器學(xué)習(xí)工作流。 ML Pipelines 提供了一套基于 DataFrame 構(gòu)建的統(tǒng)一的高級(jí) API ，可幫助用戶創(chuàng)建和調(diào)整實(shí)用的機(jī)器學(xué)習(xí)流程。

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

首先了解 ML Pipelines 內(nèi)幾個(gè)重要組件。

DataFrame

DataFrame 讓 Spark 具備了處理大規(guī)模結(jié)構(gòu)化數(shù)據(jù)的能力。

 

 

RDD 是分布式 Java 對象的集合，對象的內(nèi)部數(shù)據(jù)結(jié)構(gòu)對于 RDD 而言不可知。DataFrame 是一種以 RDD 為基礎(chǔ)的分布式數(shù)據(jù)集，RDD 中存儲(chǔ)了 Row 對象，Row 對象提供了詳細(xì)的結(jié)構(gòu)信息，即模式(schema)，使得 DataFrame 具備了結(jié)構(gòu)化數(shù)據(jù)的能力。

Transforme

Transformer 通常是一個(gè)數(shù)據(jù)/特征變換的類，或一個(gè)訓(xùn)練好的模型。

每個(gè) Transformer 都有 transform 函數(shù)，用于將一個(gè) DataFrame 轉(zhuǎn)換為另一個(gè) DataFrame 。一般 transform 的過程是在輸入的 DataFrame 上添加一列或者多列 ，Transformer.transform也是惰性執(zhí)行，只會(huì)生成新的 DataFrame 變量，而不會(huì)去提交 job 計(jì)算 DataFrame 中的內(nèi)容。

 

 

Estimator

Estimator 抽象了從輸入數(shù)據(jù)學(xué)習(xí)模型的過程，每個(gè) Estimator 都實(shí)現(xiàn)了 fit 方法，用于給定 DataFrame 和 Params 后，生成一個(gè) Transformer(即訓(xùn)練好的模型)，每當(dāng)調(diào)用 Estimator.fit() 后，都會(huì)產(chǎn)生 job 去訓(xùn)練模型，得到模型參數(shù)。

Param

可以通過設(shè)置 Transformer 或 Estimator 實(shí)例的參數(shù)來設(shè)置模型參數(shù)，也可以通過傳入 ParamMap 對象來設(shè)置模型參數(shù)。

 

 

Pipeline

Pipeline 定義了一組數(shù)據(jù)處理流程，可以在 Pipeline 中加入 Transformer、Estimator 或另一個(gè) Pipeline。Pipeline 繼承自 Estimator，調(diào)用 Pipeline.fit 方法后返回一個(gè) Transformer——PipelineModel;PipelineModel 繼承自 Transformer，用于將輸入經(jīng)過 Pipeline 的各個(gè) Transformer 的變換后，得到最終輸出。

Spark MLlib 典型流程如下：

• 構(gòu)造訓(xùn)練數(shù)據(jù)集

• 構(gòu)建各個(gè) Stage

• Stage 組成 Pipeline

• 啟動(dòng)模型訓(xùn)練

• 評估模型效果

• 計(jì)算預(yù)測結(jié)果

通過一個(gè) 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")
}

模型選擇與調(diào)參

Spark MLlib 提供了 CrossValidator 和 TrainValidationSplit 兩個(gè)模型選擇和調(diào)參工具。模型選擇與調(diào)參的三個(gè)基本組件分別是 Estimator、ParamGrid 和 Evaluator，其中 Estimator 包括算法或者 Pipeline;ParamGrid 即 ParamMap 集合，提供參數(shù)搜索空間;Evaluator 即評價(jià)指標(biāo)。

CrossValidator

 

 

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

CrossValidator 將數(shù)據(jù)集按照交叉驗(yàn)證數(shù)切分成 n 份，每次用 n-1 份作為訓(xùn)練集，剩余的作為測試集，訓(xùn)練并評估模型，重復(fù) n 次，得到 n 個(gè)評估結(jié)果，求 n 次的平均值作為這次交叉驗(yàn)證的結(jié)果。接著對每個(gè)候選 ParamMap 重復(fù)上面的過程，選擇最優(yōu)的 ParamMap 并重新訓(xùn)練模型，得到最優(yōu)參數(shù)的模型輸出。

 

舉個(gè)例子：

 

TrainValidationSplit

 

TrainValidationSplit 使用 trainRatio 參數(shù)將訓(xùn)練集按照比例切分成訓(xùn)練和驗(yàn)證集，其中 trainRatio 比例的樣本用于訓(xùn)練，剩余樣本用于驗(yàn)證。

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

舉個(gè)例子：

 

實(shí)現(xiàn)自定義 Transformer

 

繼承自 Transformer 類，實(shí)現(xiàn) transform 方法，通常是在輸入的 DataFrame 上添加一列或多列。

 

 

對于單輸入列，單輸出列的 Transformer 可以繼承自 UnaryTransformer 類，并實(shí)現(xiàn)其中的 createTransformFunc 方法，實(shí)現(xiàn)對輸入列每一行的處理，并返回相應(yīng)的輸出。

 

 

/ 自研機(jī)器學(xué)習(xí)框架 /

機(jī)器學(xué)習(xí)技術(shù)日新月異，卻缺少高效靈活的框架降低新技術(shù)的調(diào)研成本，而經(jīng)驗(yàn)與技術(shù)往往需要通過框架和工具來沉淀，并且算法人員常常受限于算力，導(dǎo)致離線證明有效的模型，因?yàn)轭A(yù)估時(shí)間復(fù)雜度過高而無法上線。

 

 

據(jù)此美圖數(shù)據(jù)技術(shù)團(tuán)隊(duì)以「開發(fā)簡單靈活的機(jī)器學(xué)習(xí)工作流，降低算法人員的新算法調(diào)研成本及工程人員的維護(hù)成本，并且提供常用的領(lǐng)域內(nèi)解決方案，將經(jīng)驗(yàn)沉淀」的目標(biāo)搭建了一套量身定制的機(jī)器學(xué)習(xí)框架用以解決上述問題，尤其是解決在推薦算法相關(guān)任務(wù)上遇到的問題。該框架總共包括 3 個(gè)組件：Spark Feature、Bamboo 與 Online Scorer。

Spark Feature：訓(xùn)練樣本生產(chǎn)

 

 

該組件主要用于訓(xùn)練樣本的生產(chǎn)，實(shí)現(xiàn)了靈活高效的樣本特征編碼，可以實(shí)現(xiàn)將任意特征集合放在同一個(gè)空間進(jìn)行編碼，不同特征集合共享編碼空間;為此我們提出了兩個(gè)概念：第一個(gè)是「域」，用于定義共享相同建模過程的一組特征;第二個(gè)是「空間」，用于定義共享相同編碼空間的一組域。

上圖示例中的「Old」展示了在沒有“域”和“空間”概念下的樣本特征編碼，所有特征從 1 開始編號(hào);「New」展示了將 age 和 gender 分別放到 age 域和 gender 域后，兩個(gè)域分別從 1 開始編碼，互不影響。

Spark Feature 最終采用 TFRecords 作為訓(xùn)練樣本的存儲(chǔ)格式。

Bamboo：模型定義與訓(xùn)練

該組件主要為了實(shí)現(xiàn)可擴(kuò)展、高效、簡單快速的模型定義與訓(xùn)練。為此，在設(shè)計(jì) Bamboo 時(shí)我們遵循以下原則：

1.layer 之間通過 tensor 進(jìn)行交互，layer 的輸入是 tensor，輸出也是 tensor;

2.為了最大限度地提高離線與在線效率，沒有采用太多高級(jí) api，如 keras，大多數(shù)模型與組件基于 Tensorflow 底層 api 開發(fā)，并且根據(jù) Tensorflow 官方的性能優(yōu)化指南對代碼進(jìn)行優(yōu)化;

3.提供 online-offline 的建�？蚣埽瑥�(fù)雜計(jì)算放到離線，在線只進(jìn)行輕量計(jì)算，使得復(fù)雜模型更易上線;

4.封裝數(shù)據(jù)加載、模型訓(xùn)練與導(dǎo)出、效果評估以及提供了各種輔助工具，用戶只需要定義前向推理網(wǎng)絡(luò)，同時(shí)封裝了大量的常用 layer，模型定義更快捷。

 

 

Online Scorer：在線預(yù)測服務(wù)

Online Scorer的目標(biāo)是提供一個(gè)統(tǒng)一，高效的在線推理服務(wù)，可以同時(shí)支持tensorflow，pytorch，xgboost等各種主流建模框架導(dǎo)出的模型。目前這塊工作還在進(jìn)行中，具體實(shí)現(xiàn)方案細(xì)節(jié)，我們放到后面的專題文章介紹。

 

 

以上就是美圖自研機(jī)器學(xué)習(xí)框架的簡要介紹，歡迎持續(xù)關(guān)注「美圖數(shù)據(jù)技術(shù)團(tuán)隊(duì)」，后續(xù)將帶來該平臺(tái)的詳細(xì)介紹。

 

 

標(biāo)簽： 代碼 搜索 網(wǎng)絡(luò)

版權(quán)申明：本站文章部分自網(wǎng)絡(luò)，如有侵權(quán)，請聯(lián)系：west999com@outlook.com
特別注意：本站所有轉(zhuǎn)載文章言論不代表本站觀點(diǎn)！
本站所提供的圖片等素材，版權(quán)歸原作者所有，如需使用，請與原作者聯(lián)系。