Gradient-Boosted Trees(GBT或者GBDT) 和 RandomForests 都屬于集成學(xué)習(xí)的范疇，相比于單個(gè)模型有限的表達(dá)能力，組合多個(gè)base model后表達(dá)能力更加豐富。

關(guān)于集成學(xué)習(xí)的理論知識(shí)，包括GBT和Random Forests的一些比較好的參考資料：

周志華教授的"Ensemble Methods: Foundations and Algorithms",系統(tǒng)的介紹了集成學(xué)習(xí)的理論及方法

Random Forests

Greedy Function Approximation: A GradientBoosting Machine

Stochastic GradientBoosting，Spark GBT實(shí)現(xiàn)所參考的算法

?

GBT和Random Forests二者的區(qū)別：

二者的理論思想在spark.mllib源碼閱讀-bagging方法中從模型的方差和偏差的角度做了一些簡要的介紹，在Spark官方文檔上也有一段關(guān)于二者優(yōu)劣的描述：

1、GBT比RandomForests的訓(xùn)練成本更高，原因在于GBT中各個(gè)模型以序列串行的方式進(jìn)行訓(xùn)練，通常我們說的并行GBT是指base model的并行訓(xùn)練，各個(gè)base model之間是無法做到并行的。而Random Forests

中各個(gè)子模型之間可以做到并行化。

2、Random Forests的base model越多越有助于降低過擬合，而GBT中base model越多會(huì)提高過擬合的程度。

3、二者訓(xùn)練的時(shí)間成本不同，因此調(diào)參的成本不同。有限的時(shí)間內(nèi)Random Forests可以實(shí)驗(yàn)更多的參數(shù)組合。

4、通常來看，Random Forests的base model會(huì)得到一棵規(guī)模適中的樹，而GBT為了降低在basemodel數(shù)量多時(shí)引發(fā)的過擬合，會(huì)限制其base model的規(guī)模。

?

下面來看看Spark中GBT的實(shí)現(xiàn)過程，主要包括3部分：GBT模型、GBT參數(shù)配置、GBT訓(xùn)練算法：

GradientBoostedTrees：
GBT的實(shí)現(xiàn)過程由GradientBoostedTrees類驅(qū)動(dòng)并向用戶暴露模型的訓(xùn)練方法。GradientBoostedTrees的2個(gè)關(guān)鍵方法是train和run，在run中，根據(jù)用戶定義的模型配置類boostingStrategy來調(diào)用ml包下的GradientBoostedTrees類進(jìn)行模型的訓(xùn)練，最后根據(jù)訓(xùn)練得到的參數(shù)來新建一個(gè)GradientBoostedTreesModel：

def train(
? ? ? ? ? ?input: RDD[LabeledPoint],
? ? ? ? ? ?boostingStrategy: BoostingStrategy): GradientBoostedTreesModel = {
? new GradientBoostedTrees(boostingStrategy, seed = 0).run(input)
? }
def run(input: RDD[LabeledPoint]): GradientBoostedTreesModel = {
? val algo = boostingStrategy.treeStrategy.algo
? //import org.apache.spark.ml.tree.impl.{GradientBoostedTrees => NewGBT}
? val (trees, treeWeights) = NewGBT.run(input.map { point =>
? ? NewLabeledPoint(point.label, point.features.asML)
? }, boostingStrategy, seed.toLong)
? new GradientBoostedTreesModel(algo, trees.map(_.toOld), treeWeights)
}
GradientBoostedTreesModel：
GradientBoostedTreesModel用來保存訓(xùn)練后的模型，其繼承自TreeEnsembleModel。各個(gè)Base model保存在trees數(shù)組中，每個(gè)base model的權(quán)重在treeWeights數(shù)組中，

其父類TreeEnsembleModel實(shí)現(xiàn)的predict方法即是對(duì)各個(gè)base model的預(yù)測(cè)值加權(quán)treeWeights 得到最終的預(yù)測(cè)值。

class GradientBoostedTreesModel @Since("1.2.0") (
? ? @Since("1.2.0") override val algo: Algo, //模型算法：分類 or 回歸
? ? @Since("1.2.0") override val trees: Array[DecisionTreeModel], //base model的數(shù)組
? ? @Since("1.2.0") override val treeWeights: Array[Double]) //每個(gè)base model的權(quán)重
? extends TreeEnsembleModel(algo, trees, treeWeights, combiningStrategy = Sum)
BoostingStrategy
GBT的配置信息類，可配置的信息包括

treeStrategy：base tree的配置信息

Loss：損失函數(shù)，默認(rèn)參數(shù)為2分類問題用LogLoss， 回歸問題用SquaredError

numIterations：GBT的迭代次數(shù)，默認(rèn)值為100

learningRate：學(xué)習(xí)速率，默認(rèn)值為0.1

validationTol：通過驗(yàn)證集判斷訓(xùn)練終止的條件：驗(yàn)證集上歷史最小的殘差 - 驗(yàn)證集當(dāng)前殘差 < validationTol*max(驗(yàn)證集當(dāng)前殘差, 0.01) 即提前終止訓(xùn)練

在訓(xùn)練GBT時(shí)，base tree的參數(shù)設(shè)置也很重要，base tree的參數(shù)由Strategy類維護(hù)，Strategy的默認(rèn)值如下，在訓(xùn)練GBT時(shí)，務(wù)必要重新設(shè)置Strategy的值，這里我對(duì)可以設(shè)定的值都做了備注，方便初次使用的同學(xué)進(jìn)行調(diào)參：

@Since("1.0.0") @BeanProperty var algo: Algo,//算法的類別：分類還是回歸 {Classification、Regression}

@Since("1.0.0") @BeanProperty var impurity: Impurity,//計(jì)算信息增益的準(zhǔn)則 分類{基尼指數(shù)、信息增益} 回歸{impurity.Variance}

@Since("1.0.0") @BeanProperty var maxDepth: Int, //樹的最大深度

@Since("1.2.0") @BeanProperty var numClasses: Int = 2,//類別數(shù)

@Since("1.0.0") @BeanProperty var maxBins: Int = 32,//連續(xù)特征離散化的分箱數(shù)

@Since("1.0.0") @BeanProperty var quantileCalculationStrategy: QuantileStrategy = Sort,//計(jì)算分裂點(diǎn)的算法，待定

@Since("1.0.0") @BeanProperty var categoricalFeaturesInfo: Map[Int, Int] = Map[Int, Int](),//存儲(chǔ)每個(gè)分類特征的值數(shù)目

@Since("1.2.0") @BeanProperty var minInstancesPerNode: Int = 1,//子結(jié)點(diǎn)擁有的最小樣本實(shí)例數(shù)，一個(gè)終止條件

@Since("1.2.0") @BeanProperty var minInfoGain: Double = 0.0,//最小的信息增益值，這個(gè)應(yīng)該是用來控制迭代終止的

@Since("1.0.0") @BeanProperty var maxMemoryInMB: Int = 256,//聚合使用的內(nèi)存大小。待定

@Since("1.2.0") @BeanProperty var subsamplingRate: Double = 1,//用于訓(xùn)練數(shù)據(jù)的抽樣率

@Since("1.2.0") @BeanProperty var useNodeIdCache: Boolean = false,//待定

@Since("1.2.0") @BeanProperty var checkpointInterval: Int = 10 //checkpoint

模型的損失函數(shù)在BoostingStrategy類中自動(dòng)設(shè)置，在二分類模型中損失函數(shù)被定義為LogLoss(對(duì)數(shù)損失函數(shù))、在回歸問題中損失函數(shù)被定義為SquaredError(平方損失函數(shù))。在Spark2.1.0版本中還沒有實(shí)現(xiàn)對(duì)多分類GBT的損失函數(shù)及多分類GBT模型。對(duì)于自定義損失函數(shù)，需要繼承org.apache.spark.mllib.tree.loss.Loss這個(gè)類，并覆寫gradient和computeError方法。

GradientBoostedTrees：
GradientBoostedTrees類是Spark訓(xùn)練GBT模型參數(shù)的類，模型的訓(xùn)練主要分為2步：1、將分類問題轉(zhuǎn)化為回歸問題，在GradientBoostedTrees的run方法中完成：

def run(
? ? input: RDD[LabeledPoint],
? ? boostingStrategy: OldBoostingStrategy,
? ? seed: Long): (Array[DecisionTreeRegressionModel], Array[Double]) = {
? val algo = boostingStrategy.treeStrategy.algo
? //都轉(zhuǎn)化為回歸問題
? algo match {
? ? case OldAlgo.Regression => GradientBoostedTrees.boost(input, input, boostingStrategy, validate = false, seed)
? ? case OldAlgo.Classification =>
? ? ? // Map labels to -1, +1 so binary classification can be treated as regression.
? ? ? val remappedInput = input.map(x => new LabeledPoint((x.label * 2) - 1, x.features))
? ? ? GradientBoostedTrees.boost(remappedInput, remappedInput, boostingStrategy, validate = false, seed)
? ? case _ => throw new IllegalArgumentException(s"$algo is not supported by gradient boosting.")
? }
}
2、問題統(tǒng)一轉(zhuǎn)化為回歸問題后，調(diào)用GradientBoostedTrees的boost進(jìn)行參數(shù)的訓(xùn)練，看一下整個(gè)訓(xùn)練過程的核心代碼(在源碼的基礎(chǔ)上有刪減)：

// Initialize gradient boosting parameters
val numIterations = boostingStrategy.numIterations //總的迭代次數(shù)，決定了生成
val baseLearners = new Array[DecisionTreeRegressionModel](numIterations) //保存每次迭代的base模型的數(shù)組
val baseLearnerWeights = new Array[Double](numIterations)//模型權(quán)重？
val loss = boostingStrategy.loss //定義的損失函數(shù)
val learningRate = boostingStrategy.learningRate
// Prepare strategy for individual trees, which use regression with variance impurity.
val treeStrategy = boostingStrategy.treeStrategy.copy
val validationTol = boostingStrategy.validationTol
treeStrategy.algo = OldAlgo.Regression //org.apache.spark.mllib.tree.configuration.{Algo => OldAlgo}
treeStrategy.impurity = OldVariance
treeStrategy.assertValid()
// Cache input
val persistedInput = if (input.getStorageLevel == StorageLevel.NONE) {
? input.persist(StorageLevel.MEMORY_AND_DISK)
? true
} else {
? false
}
// Prepare periodic checkpointers 定期Checkpointer
val predErrorCheckpointer = new PeriodicRDDCheckpointer[(Double, Double)](
? treeStrategy.getCheckpointInterval, input.sparkContext)
val validatePredErrorCheckpointer = new PeriodicRDDCheckpointer[(Double, Double)](
? treeStrategy.getCheckpointInterval, input.sparkContext)
?
val firstTree = new DecisionTreeRegressor().setSeed(seed)
//實(shí)際是用隨機(jī)森林訓(xùn)練的一棵樹,GBT中樹的深度通常較小
//RandomForest.run(data, oldStrategy, numTrees = 1, featureSubsetStrategy = "all", seed = $(seed), instr = Some(instr), parentUID = Some(uid))
val firstTreeModel = firstTree.train(input, treeStrategy)
val firstTreeWeight = 1.0
baseLearners(0) = firstTreeModel
baseLearnerWeights(0) = firstTreeWeight
//(預(yù)測(cè)值，誤差值)
//如改成多分類的話應(yīng)該是(list<pred>, list<Error>) 即每棵樹的預(yù)測(cè)值和誤差值
var predError: RDD[(Double, Double)] =
? computeInitialPredictionAndError(input, firstTreeWeight, firstTreeModel, loss)
predErrorCheckpointer.update(predError)
var validatePredError: RDD[(Double, Double)] =
? computeInitialPredictionAndError(validationInput, firstTreeWeight, firstTreeModel, loss)
if (validate) validatePredErrorCheckpointer.update(validatePredError)
var bestValidateError = if (validate) validatePredError.values.mean() else 0.0
var bestM = 1
var m = 1
var doneLearning = false
while (m < numIterations && !doneLearning) {
? // Update data with pseudo-residuals
? //predError　(預(yù)測(cè)值，誤差值) 預(yù)測(cè)值是前m-1輪的預(yù)測(cè)值之和，誤差值為lable-預(yù)測(cè)值
? //如改成多分類的話 此時(shí)該樣本的loss即可以用logitloss來表示，并對(duì)f1~fk都可以算出一個(gè)梯度，f1~fk便可以計(jì)算出當(dāng)前輪的殘差，供下一輪迭代學(xué)習(xí)。
? val data = predError.zip(input).map { case ((pred, _), point) =>
? ? LabeledPoint(-loss.gradient(pred, point.label), point.features)//
? }
? val dt = new DecisionTreeRegressor().setSeed(seed + m)
? val model = dt.train(data, treeStrategy)//訓(xùn)練下一個(gè)base model
? // Update partial model
? baseLearners(m) = model
? // Note: The setting of baseLearnerWeights is incorrect for losses other than SquaredError.
? // ? ? ? Technically, the weight should be optimized for the particular loss.
? // ? ? ? However, the behavior should be reasonable, though not optimal.
? // 這里learningRate是一個(gè)固定值，沒有使用shrinkage技術(shù)
? baseLearnerWeights(m) = learningRate // learningRate同時(shí)作為model的權(quán)重
? predError = updatePredictionError(
? ? input, predError, baseLearnerWeights(m), baseLearners(m), loss)
? predErrorCheckpointer.update(predError)
? if (validate) {//驗(yàn)證集，驗(yàn)證是否提前終止訓(xùn)練
? ? // Stop training early if
? ? // 1. Reduction in error is less than the validationTol or
? ? // 2. If the error increases, that is if the model is overfit.
? ? // We want the model returned corresponding to the best validation error.
? ? validatePredError = updatePredictionError(
? ? ? validationInput, validatePredError, baseLearnerWeights(m), baseLearners(m), loss)
? ? validatePredErrorCheckpointer.update(validatePredError)
? ? val currentValidateError = validatePredError.values.mean()
? ? if (bestValidateError - currentValidateError < validationTol * Math.max(
? ? ? currentValidateError, 0.01)) {
? ? ? doneLearning = true
? ? } else if (currentValidateError < bestValidateError) {
? ? ? bestValidateError = currentValidateError
? ? ? bestM = m + 1
? ? }
? }
? m += 1
}
GBT的訓(xùn)練是一個(gè)串行的過程，base treemodel在前一輪迭代殘差的基礎(chǔ)上逐棵生成。每次生成一棵樹之后需要更新整個(gè)數(shù)據(jù)集的殘差，再進(jìn)行下一輪的訓(xùn)練。在數(shù)據(jù)集規(guī)模較大，并且迭代輪次比較多時(shí)，訓(xùn)練比較耗時(shí)，這在一定程度上增加了模型調(diào)參的成本。

?

截至Spark2.0.0，Spark的GBT模型比較初級(jí)，在分類問題上目前只支持2分類問題，梯度下降的過程控制也比較簡單，難于適應(yīng)一些精度要求高的的機(jī)器學(xué)習(xí)任務(wù)，因此目前版本下的Spark來做GBT模型并不是一個(gè)好的選擇。相比較而言，XGBOOST是一個(gè)更好的選擇，當(dāng)然，有條件的情況下順著Spark GBT的思路做一些改進(jìn)也能達(dá)到接近的效果。
?

創(chuàng)作挑戰(zhàn)賽新人創(chuàng)作獎(jiǎng)勵(lì)來咯，堅(jiān)持創(chuàng)作打卡瓜分現(xiàn)金大獎(jiǎng)

總結(jié)

以上是生活随笔為你收集整理的spark.mllib源码阅读：GradientBoostedTrees的全部內(nèi)容，希望文章能夠幫你解決所遇到的問題。

如果覺得生活随笔網(wǎng)站內(nèi)容還不錯(cuò)，歡迎將生活随笔推薦給好友。