? ? ??

本文將重點介紹XGBoost基于Spark平臺Scala版本的實現，帶領大家逐步完成特征提取、變換和選擇、XGBoost模型訓練、Pipelines、模型選擇。

XGBoost回顧

? ? ? ?

? ? ? ?XGBoost（Extreme Gradient Boosting）由華盛頓大學的陳天奇博士提出，最開始作為分布式（深度）機器學習研究社區（DMLC）小組的研究項目之一。后因在希格斯（Higgs）機器學習挑戰賽中大放異彩，被業界所熟知，在數據科學應用中廣泛應用。目前，一些主流的互聯網公司如騰訊、阿里巴巴等都已將XGBoost應用到其業務中，在各種數據科學競賽中XGBoost也成為競賽者們奪冠的利器。XGBoost在推薦、搜索排序、用戶行為預測、點擊率預測、產品分類等問題上取得了良好的效果。雖然這些年神經網絡（尤其是深度神經網絡）變得越來越流行，但XGBoost仍舊在訓練樣本有限、訓練時間短、、調參知識缺乏的場景下具有獨特的優勢。相比深度神經網絡，XGBoost能夠更好地處理表格數據，并具有更強的可解釋性，另外具有易于調參、輸入數據不變性等優勢。

? ? ? ?XGBoost是Gradient Boosting的實現，相比其他實現方法，XGBoost做了很多優化，在模型訓練速度和精度上都有明顯提升，其優良特性如下。

1）將正則項加入目標函數中，控制模型的復雜度，防止過擬合。

2）對目標函數進行二階泰勒展開，同時用到了一階導數和二階導數。

3）實現了可并行的近似直方圖算法。

4）實現了縮減和列采樣（借鑒了GBDT和隨機森林）。

5）實現了快速直方圖算法，引入了基于loss-guide的樹構建方法（借鑒了LightGBM）。

6）實現了求解帶權值的分位數近似算法（weighted quantile sketch）。

7）可根據樣本自動學習缺失值的分裂方向，進行缺失值處理。

8）數據預先排序，并以塊（block）的形式保存，有利于并行計算。

9）采用緩存感知訪問、外存塊計算等方式提高數據訪問和計算效率。

10）基于Rabit實現分布式計算，并集成于主流大數據平臺中。

11）除CART作為基分類器外，XGBoost還支持線性分類器及LambdaMART排序模型等算法。

12）實現了DART，引入Dropout技術。

? ? ? ?目前已經有越來越多的開發人員為XGBoost開源社區做出了貢獻。XGBoost實現了多種語言的包，如Python、Scala、Java等。Python用戶可將XGBoost與scikit-learn集成，實現更為高效的機器學習應用。另外，XGBoost集成到了Spark、Flink等主流大數據平臺中。

分布式XGBoost

? ? ? ?也許在競賽中我們很少或者從不使用分布式XGBoost版本，可是在工業界爆炸式增長的數據規模，單機模式是很難滿足用戶需求，XGBoost也相應推出了分布式版本，這也是XGBoost如此流行的重要原因。本文將重點介紹XGBoost基于Spark平臺的實現，帶領大家逐步完成Spark版本的特征提取、變換和選擇，以及XGBoost模型訓練、Pipelines、模型選擇。

1.基于Spark平臺實現

? ? ? Spark是一個通用且高效的大數據處理引擎，它是基于內存的大數據并行計算框架。因為Spark計算基于內存，因此能夠保證大數據計算的實時性，相比傳統的Hadoop MapReduce效率提升很多。Spark擁有一個豐富的生態環境，以Spark為核心，涵蓋支持：結構化數據查詢與分析的Spark SQL、分布式機器學習庫MLlib、并行圖計算框架GraphX、可容錯流計算框架Spark Streaming等。由于Spark在工業界廣泛應用，用戶群體龐大，因此XGBoost推出了XGBoost4J-Spark以支持Spark平臺。

1.1?Spark架構

如圖1所示，Spark主要由如下組件構成。

• Client：提交Spark job的客戶端。

• Driver：接受Spark job請求，啟動SparkContext。

• SparkContext：整個應用的上下文，可以控制應用的生命周期。

• ClusterManager：集群管理器，為Application分配資源，包括多種類型，如Spark自帶的Standalone、Meso或者YARN等。

• Worker：集群中任意可執行Application代碼的節點，運行一個或者多個Executor。

• Executor：在Worker節點中提交Application的進程，啟動并運行任務，負責將數據存于內存或者硬盤中。每個Application均有各自的Executor執行任務。

? ? ? ?由圖1可知，Spark作業提交流程如下：首先Client提交應用，Driver接收到請求后，啟動SparkContext。SparkContext連接ClusterManager，ClusterManager負責為應用分配資源。Spark將在集群節點中獲取到執行任務的Executor，這些Executor負責執行計算和存儲數據。Spark將應用程序的代碼發送給Executor，最后SparkContext將任務分配給Executor去執行。

圖1 Spark結構

? ? ? ?在Spark應用中，整個執行流程在邏輯上會轉化為RDD（Resilient Distributed Dataset，彈性分布式數據集）的DAG（Directed Acyclic Graph，有向無環圖）。RDD是Spark的基本運算單元，后續會詳細介紹。Spark將任務轉化為DAG形式的工作流進行調度，并進行分布式分發。圖2通過示例展示了Spark執行DAG的整個流程。

圖2?Spark執行DAG的整個流程

? ? ? ?在圖2中，Transformations是RDD的一類操作，包括map、flatMap、filter等，該類操作是延遲執行的，即從一個RDD轉化為另一個RDD不立即執行，而只是將操作記錄下來，直到遇到Actions類的操作才會真正啟動計算過程進行計算。Actions類操作會返回結果或將RDD數據寫入存儲系統，是觸發Spark啟動計算的動因。Action算子觸發后，將所有記錄的算子生成一個RDD，Spark根據RDD之間的依賴關系將任務切分為不同的階段（stage），然后由調度器調度RDD中的任務進行計算。圖2中的A～E分別代表不同的RDD，RDD中的方塊代表不同的分區。Spark首先通過HDFS將數據讀入內存，形成RDD A和RDD C。RDD A轉化為RDD B，RDD C執行map操作轉化為RDD D，RDD B和RDD E執行join操作轉化為RDD F。RDD B和RDD E連接轉化為RDD F的過程中會執行Shuffle操作，最后RDD F通過函數saveAsSequenceFile輸出并保存到HDFS上。

1.2?RDD

? ? ??

? ? ? ?Spark引入了RDD概念，RDD是分布式內存數據的抽象，是一個容錯的、并行的數據結構，是Spark中基本的數據結構，所有計算均基于該結構進行，Spark通過RDD和RDD操作設計上層算法。

? ? ? RDD作為數據結構，本質上是一個只讀的分區記錄的集合，邏輯上可以把它想象成一個分布式數組，數組中的元素可以為任意的數據結構。一個RDD可以包含多個分區，每個分區都是數據集的一個子集。RDD可以相互依賴，通過依賴關系形成Spark的調度順序，通過RDD的操作形成整個Spark程序。

RDD有兩種操作算子：轉換（transformation）與行動（actions）。

1. 轉換

轉換操作是延遲執行的，即從一個RDD轉化為另一個RDD，且不立即執行，而只是將操作記錄下來，直到遇到Actions類的操作才會真正啟動計算過程。轉換操作包括map、flatMap、mapPartitions等多種操作，下面對常用的轉換操作進行介紹。

• map：對原始RDD中的每個元素執行一個用戶自定義函數生成一個新的RDD。任何原始RDD中的元素在新的RDD中有且只有一個元素與之對應。

• flatMap：與map類似，原始RDD中的元素通過函數生成新的元素，并將生成的RDD的每個集合中的元素合并為一個集合。

• mapPartitions：獲取每個分區的迭代器，在函數中對整個迭代器的元素（即整個分區的元素）進行操作。

• union：將兩個RDD合并，合并后不進行去重操作，保留所有元素。使用該操作的前提是需要保證RDD元素的數據類型相同。

• filter：對元素進行過濾，對每個元素應用函數，返回值為True的元素被保留。

• sample：對RDD中的元素進行采樣，獲取所有元素的子集。

• cache：將RDD元素從磁盤緩存到內存，相當于persist（MEMORY_ONLY）。

• persist：對RDD數據進行緩存，由參數StorageLevel決定數據緩存到哪里，如DISK_ONLY表示僅磁盤緩存、MEMORY_AND_DISK表示內存和磁盤均緩存等。

• groupBy：將RDD中元素通過函數生成相應的key，然后通過key對元素進行分組。

• reduceByKey：將數據中每個key對應的多個value進行用戶自定義的規約操作。

• join：相當于SQL中的內連接，返回兩個RDD以key作為連接條件的內連接。

2. 行動

? ? ? ?行動操作會返回結果或將RDD數據寫入存儲系統，是觸發Spark啟動計算的動因。行動操作包括foreach、collect等。下面對常用的行動操作進行介紹。

• foreach：對RDD中每個元素都調用用戶自定義函數操作，返回Unit。

• collect：對于分布式RDD，返回一個scala中的Array數組。

• count：返回RDD中元素的個數。

• saveAsTextFile：將數據以文本的形式存儲到HDFS的指定目錄。

? ? ? ?DataSet是分布式的數據集合，它是在Spark 1.6之后新增的一個接口，其不但具有RDD的優點，而且同時具有Spark SQL優化執行引擎的優勢。DataFrame是一個具有列名的分布式數據集，可以近似看作關系數據庫中的表，但DataFrame可以從多種數據源進行構建，如結構化數據文件、Hive中的表、RDD等。DataFrame API可以在Scala、Java、Python和R中使用。下面只介紹幾個常用的API（更多API可以參考相關資料[插圖]）。

• select（cols:Column*）：選取滿足表達式的列，返回一個新的DataFrame。其中，cols為列名或表達式的列表。

• filter（condition:Column）：通過給定條件過濾行。

• count（）：返回DataFrame行數。

• describe（cols:String*）：計算數值型列的統計信息，包括數量、均值、標準差、最小值、最大值。

• groupBy（cols:Column*）：通過指定列進行分組，分組后可通過聚合函數對數據進行聚合。

• join（right:Dataset[_]）：和另一個DataFrame進行join操作。

• withColumn（colName:String,col:Column）：添加列或者替換具有相同名字的列，返回新的DataFrame。

1.3?XGBoost4J-Spark

? ? ?

? ? ? 隨著Spark在工業界的廣泛應用，積累了大量的用戶，越來越多的企業以Spark為核心構建自己的數據平臺來支持挖掘分析類計算、交互式實時查詢計算，于是XGBoost4J-Spark應運而生。本節將介紹如何通過Spark實現機器學習，如何將XGBoost4J-Spark很好地應用于Spark機器學習處理的流水線中。

? ? ? XGBoost4J-Spark在jvm-package中實現，因此在工程中調用XGBoost4J時，只需在pom.xml文件中加入如下依賴即可：

<dependency><groupId>ml.dmlc</groupId><artifactId>xgboost4j-spark</artifactId><version>0.7</version> </dependency>

? ? ? ?圖3展示了如何將XGBoost4J-Spark應用于Spark機器學習處理的流水線框架中。首先通過Spark將數據加載為RDD、DataFrame或DataSet。如果加載類型為DataFrame/DataSet，則可通過Spark SQL對其進行進一步處理，如去掉某些指定的列等。由Spark MLlib庫完成特征工程，其提供了多種特征工程的方法供用戶選擇，此步驟是機器學習過程中非常重要的一步，因為好的特征可以決定機器學習的上限。特征工程完成后，便可將生成的訓練數據送入XGBoost4J-Spark中進行訓練，在此過程中可通過Spark MLlib進行參數調優，得到最優模型。得到訓練模型后對預測集進行預測，最終得到預測結果。為了避免每次重復的訓練模型，可將訓練好的模型保存下來，在使用時直接加載即可。另外，訓練完成后，XGBoost4J-Spark可對特征重要程度進行排名。最后，形成數據產品應用于相關業務。

圖3?XGBoost4J-Spark模型訓練流程圖

? ? ? ?0.70版本及以上版本的XGBoost4J-Spark支持用戶在Spark中使用低級和高級內存抽象，即RDD和DataFrame/DataSet，而低版本（0.60版本）的僅支持RDD方式。DataFrame/DataSet可以近似看作數據庫的一張表，不但包含數據，而且包含表結構，是結構化的數據。用戶可以方便地利用Spark提供的DataFrame/DataSet API對其操作，也可以通過用戶自定義函數（UDF）進行處理，例如，通過select函數可以很方便地選取需要的特征形成一個新的DataFrame/DataSet。以下示例將結構化數據保存在JSON文件中，并通過Spark的API解析為DataFrame，并以兩行Scala代碼來訓練XGBoost模型。

1.val df = spark.read.json("data.json") 2.//調用 XGBoost API 訓練DataFrame類型的訓練集 3.val xgboostModel = XGBoost.trainWithDataFrame( 4. df, paramMap, numRound, nWorkers, useExternalMemory)

上述代碼是XGBoost4J-Spark 0.7x版本的實現代碼，XGBoost4J-Spark 0.8x及以上版本中的部分API有所改動。訓練代碼如下：

1.val xgbClassifier = new XGBoostClassifier(paramMap). 2. setFeaturesCol("features"). 3. setLabelCol("label") 4.val xgbClassificationModel = xgbClassifier.fit(df)

下面通過示例簡單介紹XGBoost4J-Spark中的一些常用API，其他可參考官方文檔。首先，加載數據集，可通過Spark進行讀取，例如外部文件加載、Spark SQL等。然后，設置模型參數，可根據具體問題及數據分布調整模型參數：

1.val paramMap = Map( 2. "eta" -> 0.1f, 3. "num_class" -> 3, 4. "max_depth" -> 3, 5. "objective" -> "multi:softmax")

模型訓練調用方式這里不再贅述，下面介紹訓練函數中各參數的含義

• trainingData：訓練集RDD。

• params：模型訓練參數。

• round：模型迭代輪數。

• nWorkers：XGBoost訓練節點個數，如果設為0，則XGBoost會將訓練集RDD的分區數作為nWorkers的數量。

• obj：用戶定義的目標函數，默認為Null。

• eval：用戶定義的評價函數，默認為Null。

• useExternalMemory：是否利用外存緩存，如果設置為True，則可以節省運行XGBoost的RAM成本。

• missing：數據集中指定為缺省值的值（注意，此處為XGBoost會將 missing值作為缺省值，在訓練之前會將missing值置為空）。

? ? ? 模型訓練完成之后，可將模型文件進行保存以供預測時使用。模型被保存為Hadoop文件，存儲于HDFS上。0.7版本通過saveModelAsHadoopFile可實現該功能，調用示例如下：

xgboostModel.saveModelAsHadoopFile("/tmp/bst.model")

0.8及以上版本直接可通過save函數實現，如下：

xgboostModel.write.overwrite().save("/tmp/bst.model")

XGBoost可以將之前訓練好的模型文件直接加載，以供使用，0.7x版本代碼如下：

val model = XGBoost.loadModelFromHadoopFile("/tmp/bst.model")

0.8及以上版本，如下：

val model = XGBoostClassificationModel.load("/tmp/bst.model")

此處為分類模型，若為回歸模型則為：

val model = XGBoostRegressionModel.load("/tmp/bst.model")

將預測集傳入訓練好的模型即可進行預測，0.7x版本對RDD類型數據預測代碼，如下：

val predicts = model.predict(test)

0.8及以上版本則直接對DataSet類型數據進行預測，如下：

val predicts = model.transform(test)

? ? ? ?Spark訓練好的模型也可以下載到本地，通過本地的XGBoost（Python、Java或Scala）加載并進行預測。這樣既可以實現模型通過分布式訓練海量樣本，提高模型的準確度，又可以通過單機調用分布式訓練的模型進行預測，提高模型預測速度。

? ? ? ?用戶不僅可以通過DataFrame/DataSet API對數據集進行操作，而且可以通過Spark提供的MLlib機器學習包對特征進行處理。MLlib是構建于Spark之上的機器學習庫，由通用的學習算法和工具類組成。通過MLlib可以方便地對特征進行提取和轉化。MLlib還提供了非常豐富的算法，包括分類、回歸、聚類、協同過濾、降維等，用戶可以根據應用場景將這些算法和XGBoost結合使用。另外，MLlib還提供了模型選擇工具，用戶可以通過API定義的自動參數搜索過程來選擇最佳模型。

特征提取、變換和選擇

? ? ? 在將訓練集送入XGBoost4J-Spark訓練之前，可以首先通過MLlib對特征進行處理，包括特征提取、變換和選擇。這是在進行模型訓練前十分重要的一步，但不是必需的，用戶可以根據應用場景進行選擇。

在MLlib中，特征提取方法主要有如下3種。

• TF-IDF：詞頻率-逆文檔頻率，是常見的文本預處理步驟。字詞的重要性隨著它在文件中出現的次數呈正比增加，但也會隨著它在語料庫中出現的頻率呈反比下降。

• Word2Vec：其將文檔中的每個單詞都映射為一個唯一且固定長度的向量。

• CountVectorizer：用向量表示文檔中每個詞出現的次數。

? ? ? 特征變換在Spark機器學習流水線中占有重要地位，廣泛應用在各種機器學習場景中。MLlib提供了多種特征變換的方法，此處只選擇常用的方法進行介紹。

（1）StringIndexer

StringIndexer將標簽的字符串列編碼為標簽索引列。索引取值為[0,numLabels]，按標簽頻率排序。如下表所示，category列為原數據列，categoryIndex列為通過StringIndexer編碼后的列。a出現最頻繁（編碼為0.0），依次為c（編碼為1.0）、b（編碼為2.0）。

調用代碼非常簡單，只需如下兩行即可實現：

1.val indexer = new StringIndexer() 2. .setInputCol("category") 3. .setOutputCol("categoryIndex") 4. 5.val indexed = indexer.fit(df).transform(df)

（2）OneHotEncoder

OneHotEncoder將一列標簽索引映射到一列二進制向量，最多只有一個單值，可以將前面StringIndexer生成的索引列轉化為向量。OneHotEncoder主要應用于類別特征上，如性別、國籍等。類別特征不能直接應用于機器學習模型中，因為即使通過StringIndexer將字符串轉為數值型特征后，模型往往默認數據是連續的，并且是有序的；但是，類別特征數字并不是有序的，只是每個數字代表一個類別。

OneHotEncoder可以結合StringIndexer使用，代碼如下：

1.val indexer = new StringIndexer() 2. .setInputCol("category") 3. .setOutputCol("categoryIndex") 4. .fit(df) 5.val indexed = indexer.transform(df) 6. 7.val encoder = new OneHotEncoder() 8. .setInputCol("categoryIndex") 9. .setOutputCol("categoryVec") 10. 11.val encoded = encoder.transform(indexed)

（3）Normalizer

Normalizer可以將多行向量輸入轉化為統一的形式。參數p（默認為2）用來指定正則化操作中使用的p-norm。正則化操作可以使輸入數據標準化并提高后期模型的效果。

1.val normalizer = new Normalizer() 2. .setInputCol("features") 3. .setOutputCol("normFeatures") 4. .setP(1.0) 5. 6.val l1NormData = normalizer.transform(dataFrame)

（4）StandardScaler

StandardScaler處理Vector數據，標準化每個特征使得其有統一的標準差及（或者）均值為零。它有如下參數：

1）withStd：默認值為真，使用統一標準差方式。

2）withMean：默認為假。這種方法將產生一個稠密輸出，所以不適用于稀疏輸入。

1.val scaler = new StandardScaler() 2. .setInputCol("features") 3. .setOutputCol("scaledFeatures") 4. .setWithStd(true) 5. .setWithMean(false) 6. 7.// 通過擬合StandardScaler計算匯總統計信息 8.val scalerModel = scaler.fit(dataFrame) 9. 10.// 標準化特征 11.val scaledData = scalerModel.transform(dataFrame)

（5）MinMaxScaler

MinMaxScaler通過重新調節大小將Vector形式的列轉換到指定的范圍內，通常為[0,1]。它的參數有以下2個。

1）min：默認為0.0，為轉換后所有特征的上邊界。

2）max：默認為1.0，為轉換后所有特征的下邊界。

1.val scaler = new MinMaxScaler() 2. .setInputCol("features") 3. .setOutputCol("scaledFeatures") 4. 5.// 計算統計信息，生成MinMaxScalerModel 6.val scalerModel = scaler.fit(dataFrame) 7. 8.// 重新縮放每個特征至[min, max]范圍 9.val scaledData = scalerModel.transform(dataFrame)

（6）SQLTransformer

SQLTransformer實現了基于SQL語句定義的特征轉換，如“SELECT...FROM__THIS__...”，其中“__THIS__”表示輸入數據集的基礎表。

1.val df = spark.createDataFrame( 2. Seq((0, 1.0, 3.0), (2, 2.0, 5.0))).toDF("id", "v1", "v2") 3. 4.val sqlTrans = new SQLTransformer().setStatement( 5. "SELECT *, (v1 + v2) AS v3, (v1 * v2) AS v4 FROM __THIS__") 6. 7.sqlTrans.transform(df)

（7）VectorAssembler

VectorAssembler將給定的列列表組合到單個向量列中。它可以將原始特征和一系列通過其他轉換器得到的特征合并為單一的特征向量，以訓練如邏輯回歸和決策樹等機器學習算法。

1.val assembler = new VectorAssembler() 2. .setInputCols(Array("hour", "mobile", "userFeatures")) 3. .setOutputCol("features") 4. 5.val output = assembler.transform(dataset)

除了以上介紹的幾種方法之外，MLlib還提供了其他特征變換方法，如用于特征分桶的Bucketizer、用于降維的PCA等，此處不再一一介紹，讀者如感興趣可查閱相關資料[插圖]，基于應用場景合理選擇相應的特征轉變換方法。

特征選擇是指通過剔除不相關或冗余的特征，從而達到減少特征個數、提高模型精確度、減少運行時間的目的。MLlib提供了如下幾種特征選擇的方法

• VectorSlicer：從特征向量中輸出一個新特征向量，該新特征向量為原特征向量的子集，在向量列中提取特征時很有用。

• RFormula：選擇由R模型公式指定的列。

• ChiSqSelector：Chi-Squared特征選擇，應用于類別特征數據。

XGBoost模型訓練

? ? ? ?

? ? ? 在進行XGBoost模型訓練前，通過MLlib對數據集進行特征提取、變換、選擇，能夠使數據集的特征更具有代表性，減少模型受到的噪聲干擾，提高模型精度。另外，選取出真正相關的特征簡化模型，協助理解數據產生的過程。下面通過示例介紹如何將MLlib的特征提取、變換、選擇與XGBoost結合起來，此處采用iris數據集。下面給出來0.8x版本的具體實現：

1.import ml.dmlc.xgboost4j.scala.spark.{TrackerConf, XGBoostClassificationModel, XGBoostClassifier, XGBoostRegressionModel, XGBoostRegressor} 2.import org.apache.spark.ml.feature.StringIndexer 3.import org.apache.spark.ml.feature.VectorAssembler 4.import org.apache.spark.sql.types.{DoubleType, StringType, StructField, StructType} 5. 6.// 讀取數據集，生成DataFrame 7.val schema = new StructType(Array( 8. StructField("sepal length", DoubleType, true), 9. StructField("sepal width", DoubleType, true), 10. StructField("petal length", DoubleType, true), 11. StructField("petal width", DoubleType, true), 12. StructField("class", StringType, true))) 13.val df = spark.read.schema(schema).csv("{HDFS PATH}/iris.txt") 14. 15.// 定義StringIndexer，將字符串類型列class轉為數值型列label 16.val indexer = new StringIndexer() 17. .setInputCol("class") 18. .setOutputCol("label") 19. 20.// 對前述定義的列進行轉換，并去掉原來的classz字段 21.val labelTransformed = indexer.fit(df).transform(df).drop("class") 22. 23.// 對特征進行vectorAssembler，生成features列 24.val vectorAssembler = new VectorAssembler(). 25. setInputCols(Array("sepal length", "sepal width", "petal length", "petal width")). 26. setOutputCol("features") 27.val xgbInput = vectorAssembler.transform(labelTransformed).select("features", "label") 28. 29.// 定義訓練參數 30.val paramMap = Map( 31. "eta" -> 0.1f, 32. "num_class" -> 3, 33. "max_depth" -> 3, 34. "objective" -> "multi:softmax", 35. "num_round" -> 10, 36. "num_workers" -> 1) 37. 38.// 訓練模型 39.val xgbClassifier = new XGBoostClassifier(paramMap).setFeaturesCol("features").setLabelCol("label") 40.val xgbClassificationModel = xgbClassifier.fit(xgbInput)

Piplelines

? ? ? ?MLlib中的Pipeline主要受scikit-learn項目的啟發，旨在更容易地將多個算法組合成單個管道或工作流，向用戶提供基于DataFrame的更高層次的API庫，以更方便地構建復雜的機器學習工作流式應用。一個Pipeline可以集成多個任務，如特征變換、模型訓練、參數設置等。下面介紹幾個重要的概念。

• DataFrame：相比于RDD，DataFrame還包含schema信息，可以將其近似看作數據庫中的表。

• Transformer：Transformer可以看作將一個DataFrame轉換成另一個DataFrame的算法。例如，模型即可看作一個Transformer，它將預測集的DataFrame轉換成了預測結果的DataFrame。

• Estimator：一種可以適應DataFrame來生成Transformer的算法，操作于DataFrame數據并生成一個Transformer。

• Pipeline：可以連接多個Transformer和Estimator形成機器學習的工作流。

• Parameter：設置Transformer和Estimator的參數。

? ? ? ?Pipeline是多個階段形成的一個序列，每個階段都是一個Transformer或者Estimator。這些階段按順序執行，當數據通過DataFrame輸入Pipeline中時，數據在每個階段按相應規則進行轉換。在Transformer階段，對DataFrame調用transform（）方法。在Estimator階段，對DataFrame調用fit（）方法產生一個Transformer，然后調用該Transformer的transform（）。

? ? ? ?MLlib允許用戶將特征提取/變換/選擇、模型訓練、數據預測等構成一個完整的Pipeline。XGBoost也可以作為Pipeline集成到Spark的機器學習工作流中。下面通過示例介紹如何將特征處理的Transformer和XGBoost結合起來構成Spark的Pipeline。0.8.x版本的實現代碼如下：

1.import ml.dmlc.xgboost4j.scala.spark.{TrackerConf, XGBoostClassificationModel, XGBoostClassifier, XGBoostRegressionModel, XGBoostRegressor} 2.import ml.dmlc.xgboost4j.scala.spark.XGBoostEstimator 3.import org.apache.spark.ml.feature.StringIndexer 4.import org.apache.spark.ml.feature.VectorAssembler 5.import org.apache.spark.sql.types.{DoubleType, StringType, StructField, StructType} 6.import org.apache.spark.ml.Pipeline 7. 8.// 讀取數據集，生成DataFrame 9.val schema = new StructType(Array( 10. StructField("sepal length", DoubleType, true), 11. StructField("sepal width", DoubleType, true), 12. StructField("petal length", DoubleType, true), 13. StructField("petal width", DoubleType, true), 14. StructField("class", StringType, true))) 15.val df = spark.read.schema(schema).csv("{HDFS PATH}/iris.txt") 16. 17.// 定義StringIndexer，將字符串類型列class轉為數值型列label 18.val indexer = new StringIndexer(). 19. setInputCol("class"). 20. setOutputCol("label") 21. 22.// 對特征進行vectorAssembler，生成features列 23.val vectorAssembler = new VectorAssembler(). 24. setInputCols(Array("sepal length", "sepal width", "petal length", "petal width")). 25. setOutputCol("features") 26. 27.// 定義訓練參數 28.val paramMap = Map( 29. "eta" -> 0.1f, 30. "num_class" -> 3, 31. "max_depth" -> 3, 32. "objective" -> "multi:softmax", 33. "num_round" -> 10, 34. "num_workers" -> 1) 35. 36.// 定義模型 37.val xgbClassifier = new XGBoostClassifier(paramMap).setFeaturesCol("features").setLabelCol("label") 38. 39.// 構建pipeline 40.val pipeline = new Pipeline().setStages(Array(indexer, vectorAssembler, xgbClassifier)) 41.val model = pipeline.fit(df) 42. 43.// 預測 44.val predict = model.transform(df)

模型選擇

? ? ??

? ? ? ?模型選擇是機器學習中非常重要的任務，即通過數據找到具體問題的最佳模型和參數，也稱超參數調整。模型選擇可以在單獨的Estimator（如邏輯回歸）中完成，也可以在包含多個算法或者其他步驟的Pipeline中完成。用戶可以一次調整整個Pipeline中的參數，而不是單獨調整Pipeline中的每一個元素。MLlib支持CrossValidator和TrainValidationSplit兩個模型選擇工具。

（1）CrossValidator

? ? ? ?即交叉驗證，將數據集劃分為若干份子集分別進行訓練和測試。例如，設置k值為3，CrossValidator將產生3組數據，每組數據中的2/3作為訓練集進行訓練，1/3作為測試集進行測試。CrossValidator計算3組數據訓練模型的評估準則的平均值。確定了最佳參數之后，CrossValidator使用最佳參數重新對整個數據集進行擬合得到最終模型。

（2）Train-Validation Split

? ? ? ?除了CrossValidator之外，MLlib還提供了Train-Validation Split用以超參數調整。和CrossValidator不同的是，Train-Validation Split只驗證1次，而非k次。Train-Validation Split的計算代價相較于CrossValidator更低，但是當訓練數據集不夠大時，結果可靠性不高。Train-Validation Split通過trainRatio參數將數據集分成兩個部分。例如，設置trainRatio=0.75，TrainValidation Split則將75%的數據用于訓練，25%的數據用于測試。

? ? ? ?模型選擇確定最佳參數是最大限度提高XGBoost模型的關鍵步驟之一。通過手工調整參數是一項費時又乏味的過程。最新版本的XGBoost4J-Spark可以通過MLlib的模型選擇工具進行參數調優，極大地提高了機器學習過程中參數調優的效率。下面通過一個示例來說明如何利用MLlib模型選擇工具對XGBoost進行參數調優。0.8x版本的實現代碼如下：

1.import org.apache.spark.ml.tuning.ParamGridBuilder 2.import org.apache.spark.ml.evaluation.MulticlassClassificationEvaluator 3.import org.apache.spark.ml.tuning.TrainValidationSplit 4. 5.// 創建xgbClassifier 6.val xgbClassifier = new XGBoostClassifier(paramMap).setFeaturesCol("features").setLabelCol("label") 7. 8.// 設定參數調優時參數的范圍 9.val paramGrid = new ParamGridBuilder(). 10. addGrid(xgbEstimator.maxDepth, Array(5, 6)). 11. addGrid(xgbEstimator.eta, Array(0.1, 0.4)). 12. build() 13. 14.// 構建TrainValidationSplit，設置trainRatio=0.8，即80%的數據用于訓練，20%的數據用于測試 15.val tv = new TrainValidationSplit(). 16. setEstimator(xgbEstimator). 17. setEvaluator(new MulticlassClassificationEvaluator().setLabelCol("label")). 18. setEstimatorParamMaps(paramGrid). 19. setTrainRatio(0.8) 20.val model = tv.fit(xgbInput)

? ? ? ?上述示例利用MLlib中的Train-Validation Split和RegressionEvaluator對XGBoost的eta和maxDepth兩個參數進行調整，選擇RegressionEvaluator定義的最小成本函數值的模型作為最佳模型。

? ? ? ?通過XGBoost4J-Spark，用戶可以構建一個基于Spark的更高效的數據處理流水線。該流水線可以很好地利用DataFrame/DataSet API對結構化數據進行處理，并且同時擁有強大的XGBoost作為機器學習模型。另外，XGBoost4J-Spark使得XGBoost和Spark MLlib無縫連接，使得特征提取/變換/選擇和參數調優工作比以前更容易。

以上內容摘自《 深入理解XGBoost 》一書并進行整理，經出版方授權發布。

往期精彩回顧適合初學者入門人工智能的路線及資料下載機器學習在線手冊深度學習在線手冊AI基礎下載（pdf更新到25集）本站qq群1003271085，加入微信群請回復“加群”獲取一折本站知識星球優惠券，請回復“知識星球”喜歡文章，點個在看

總結

以上是生活随笔為你收集整理的深入理解XGBoost：分布式实现的全部內容，希望文章能夠幫你解決所遇到的問題。

如果覺得生活随笔網站內容還不錯，歡迎將生活随笔推薦給好友。