Spark MLib 每周一算法 —— GBTs

2019-03-16

GBTs(Gradient-boosted trees)

GBTs也是一种基于基础决策树(DecisionTree)的集成模型(Ensemble)，它通过迭代方式不断基于当前的决策树生成下一棵树，最终将这一系列决策树的结果组合产生最终的结果，从而进行分类或者回归。和RandomForests一样，集成模型的好处是可以非常好的避免过拟合(overfitting)问题，同时它也能使得一个性能比较差的弱决策树性能逐步变好。

Spark中根据GBTs的用法，实现了2个Estimator类，分别是用作分类(处理label是类别，Spark2.4.0版本只支持二分类(binary classification))的GBTClassifier和用作回归(处理label是连续数值)的GBTRegressor，从而两者也在fit训练数据后产出不同的模型对象，GBTClassificationModel和GBTRegressionModel。两者的区别主要在于产生下一棵树时使用的目标损失函数(Loss)。数据集被模型对象transform后的输出都包含了最终预测的分类或数值(prediction)，另外GBTClassificationModel额外输出各个基础决策树得到的分类结果的加权和(rawPrediction)以及这个数据在信号函数作用后的归一化值(probability)。

伪算法过程

Bagging 和 Boosting
集成模型通常有2种：Bagging 和 Boosting。Bagging的思想是并行的用不同的策略生成多个基础模型，然后在投票阶段进行结果组合。而Boosting的思想是从前一个基础模型的准确率出发，调整测试数据的权重或者优化目标，从而生成下一个进化的基础模型，最终按照这些基础模型的权重得出最终结果。GBTs是一种Boosting算法，它的迭代依据是之前所有迭代产生的决策树的预测结果与标签数据的残差函数，下一棵树将以这个残差函数的导数作为新的label进行优化，所以新生成的树是整个预测的残差梯度方向的优化，即被称为梯度提升(Gradient-boosted)。最终结果为所有这一系列树结果的加权和函数。

迭代中使用的损失函数
在二分类时，迭代中产生下棵决策树的损失函数如下：
$$2\sum_{i=1}^{N}log(1+e^{-2y_{i}F(x_{i})})$$
其中N为训练数据样本个数，$y_{i}$为第i个训练数据的实际分类值(这里取-1或1)，$F(x_{i})$为当前决策树对第i条数据的预测分类值(也是取-1或1)。所以当预测正确时，残差值为$2log(1+e^{-2})$，不正确时为$2log(1+e^{-2})$。下一棵决策树将依据该函数的导函数 $\frac{-4y_{i}}{1+e^{2y_{i}F(x_{i})}}$ 作为新的label值进行生成，所以迭代过程使用的决策树都是回归树。

在回归时，迭代中产生下棵决策树的损失函数有以下2个选择：
平方误差：$$\sum_{i=1}^{N}(y_{i}-F(x_{i}))^{2}$$
和绝对值误差：$$\sum_{i=1}^{N}|y_{i}-F(x_{i})|$$

迭代提前终止
树的迭代次数增多，能有效提升精度，但如果树的数量过多，那就会出现过拟合问题。Spark中可以设置在每次迭代后进行数据验证，当某个迭代的结果(当前残差)对之前最佳的残差的提升在验证数据(由validationIndicatorCol标明)上不足某个阈值(validationTol*max(当前残差,0.01))时，整个学习过程就会终止。

参数

GBTs会在计算过程中使用DecisionTree算法，所以很多DecisionTree中使用的参数在这里也适用，就不一一列出了。

输入输出相关:

validationIndicatorCol: 输入训练数据中表示是否是验证数据的字段名称，数据值是False代表非验证数据，True代表是验证数据。这里的验证数据仅用于迭代提前终止的验证，参考上面的迭代提前终止部分的介绍 (默认值: 无)
GBTClassifier.rawPredictionCol: 输出结果数据中存放一系列决策树结果值的加权和的字段名称 (默认值: rawPrediction)
GBTClassifier.probabilityCol: 输出结果数据中存放rawPredictionCol按信号函数做归一化值的字段名称 (默认值: probability)
GBTClassifier.predictionCol: 输出结果数据中最终判别类别的字段名称 (默认值: prediction)
GBTRegressor.predictionCol: 输出结果数据中最终预测的数据值的字段名称 (默认值: prediction)

算法模型相关:

GBTClassifier.lossType: 迭代时的损失函数(默认值：logistic)：
可选项1. logistic-分类时只能选择这个函数，具体公式见上面算法部分。
GBTRegressor.lossType: 迭代时的损失函数(默认值：squared)：
可选项1. squared-回归时可选择平方函数，具体公式见上面算法部分。
可选项2. absolute-回归时可选择绝对值函数，具体公式见上面算法部分。
maxIter: 最大迭代次数，每迭代一次就生成一棵决策树，提高这个值能提升模型性能，但模型生成时间就会线性增长。(>=0整数，默认值: 20)
stepSize: 初始决策树(初始决策树权重为1.0)之后的每次迭代树的权重，一般不建议修改这个值，如果迭代无法稳定收敛，可以适当降低这个参数值。(>=0实数，默认值: 0.1)
subsamplingRate: 随机抽样率，生成每棵决策树时选取的测试数据集的抽样子集，子集越小，模型生成速度越快。建议使用全集，不做抽样。((0,1]之间的实数，默认值: 1.0)
validationTol: 迭代残差提升阈值，参考上面的迭代提前终止部分的介绍，比如设置为1.0，则表示每次残差的提升至少要一倍以上，不然就会终止学习。(>0实数，默认值: 0.01)

model对象的成员

numTrees: GBTs生成的一系列树的个数。
totalNumNodes: GBTs生成的一系列树包含的节点总数，包括所有决策树的内部节点和叶子节点。
trees: GBTs生成的一系列树模型对象的集合，类型都是DecisionTreeRegressionModel的数组。即使是分类，每一颗迭代生成的也是回归树。
treeWeights: GBTs生成的一系列树中每棵树的权重。
toDebugString: 输出整个系列树的树形结构。

例子

//======================= classifier =======================
    var training = spark.createDataFrame(Seq(
      (1.0, Vectors.dense(0.0, 11.0, 1.0)),
      (0.0, Vectors.dense(2.0, 10.0, 1.0)),
      (0.0, Vectors.dense(2.0, 13.0, 1.0)),
      (1.0, Vectors.dense(0.0, 12.0, 5.0))
    )).toDF("label", "features")
    training.show(false)
    //    +-----+--------------+
    //    |label|features      |
    //    +-----+--------------+
    //    |1.0  |[0.0,11.0,1.0]|
    //    |0.0  |[2.0,10.0,1.0]|
    //    |0.0  |[2.0,13.0,1.0]|
    //    |1.0  |[0.0,12.0,5.0]|
    //    +-----+--------------+

    val gc = new GBTClassifier().setMaxIter(5)

    val cModel = gc.fit(training)

    println(s"GBTClassifier model's trees and totalNodes: ${cModel.numTrees} / ${cModel.totalNumNodes}")
    //  5 / 15  (5棵树，15个节点)
    println("GBTClassifier model's treeWeights: " + cModel.treeWeights.mkString(", "))
    //  [1.0, 0.1, 0.1, 0.1, 0.1]
    println("GBTClassifier model's debug info: " + cModel.toDebugString)
    //    Tree 0 (weight 1.0):
    //    If (feature 0 <= 1.0)
    //    Predict: 1.0
    //    Else (feature 0 > 1.0)
    //    Predict: -1.0
    //    Tree 1 (weight 0.1):
    //    If (feature 0 <= 1.0)
    //    Predict: 0.4768116880884702
    //    Else (feature 0 > 1.0)
    //    Predict: -0.4768116880884702
    //    Tree 2 (weight 0.1):
    //    If (feature 0 <= 1.0)
    //    Predict: 0.4381935810427206
    //    Else (feature 0 > 1.0)
    //    Predict: -0.4381935810427206
    //    Tree 3 (weight 0.1):
    //    If (feature 0 <= 1.0)
    //    Predict: 0.4051496802845983
    //    Else (feature 0 > 1.0)
    //    Predict: -0.4051496802845983
    //    Tree 4 (weight 0.1):
    //    If (feature 0 <= 1.0)
    //    Predict: 0.3765841318352991
    //    Else (feature 0 > 1.0)
    //    Predict: -0.3765841318352991

    var data = Array(
      Vectors.dense(1.0, 15.0, 13),
      Vectors.dense(3.0, 2.0, 1),
      Vectors.dense(0.0, 22.0, 5)
    )
    var test = spark.createDataFrame(data.map(Tuple1.apply)).toDF("features")
    test.show
    //    +---------------+
    //    |       features|
    //    +---------------+
    //    |[1.0,15.0,13.0]|
    //    |  [3.0,2.0,1.0]|
    //    | [0.0,22.0,5.0]|
    //    +---------------+

    cModel.transform(test).show(false)
    //    +---------------+----------------------------------------+----------------------------------------+----------+
    //    |features       |rawPrediction                           |probability                             |prediction|
    //    +---------------+----------------------------------------+----------------------------------------+----------+
    //    |[1.0,15.0,13.0]|[-1.1696739081251089,1.1696739081251089]|[0.08791619739525251,0.9120838026047475]|1.0       |
    //    |[3.0,2.0,1.0]  |[1.1696739081251089,-1.1696739081251089]|[0.9120838026047474,0.08791619739525258]|0.0       |
    //    |[0.0,22.0,5.0] |[-1.1696739081251089,1.1696739081251089]|[0.08791619739525251,0.9120838026047475]|1.0       |
    //    +---------------+----------------------------------------+----------------------------------------+----------+



//======================= regression =======================
    training = spark.createDataFrame(Seq(
      (3.0, Vectors.dense(0.0, 1.1, 0.1)),
      (0.0, Vectors.dense(2.0, 1.0, -1.0)),
      (0.0, Vectors.dense(2.0, 1.3, 1.0)),
      (1.0, Vectors.dense(0.0, 1.2, -0.5))
    )).toDF("label", "features")
    training.show(false)
    //    +-----+--------------+
    //    |label|features      |
    //    +-----+--------------+
    //    |3.0  |[0.0,1.1,0.1] |
    //    |0.0  |[2.0,1.0,-1.0]|
    //    |0.0  |[2.0,1.3,1.0] |
    //    |1.0  |[0.0,1.2,-0.5]|
    //    +-----+--------------+

    val gr = new GBTRegressor().setMaxIter(5)

    val rModel = gr.fit(training)

    println(s"GBTRegressor model's trees and totalNodes: ${cModel.numTrees} / ${rModel.totalNumNodes}")
    //  5 / 9  (5棵树，9个节点)
    println("GBTRegressor model's treeWeights: " + rModel.treeWeights.mkString(", "))
    //  [1.0, 0.1, 0.1, 0.1, 0.1]
    println("GBTRegressor model's debug info: " + rModel.toDebugString)
    //    Tree 0 (weight 1.0):
    //    If (feature 0 <= 1.0)
    //    If (feature 1 <= 1.15)
    //    Predict: 3.0
    //    Else (feature 1 > 1.15)
    //    Predict: 1.0
    //    Else (feature 0 > 1.0)
    //    Predict: 0.0
    //    Tree 1 (weight 0.1):
    //    Predict: 0.0
    //    Tree 2 (weight 0.1):
    //    Predict: 0.0
    //    Tree 3 (weight 0.1):
    //    Predict: 0.0
    //    Tree 4 (weight 0.1):
    //    Predict: 0.0
    //  后几颗树在没有残差的情况下就等于不用生成了

    data = Array(
      Vectors.dense(-1.0, 1.5, 1.3),
      Vectors.dense(3.0, 2.0, -0.1),
      Vectors.dense(0.0, 2.2, -1.5)
    )
    test = spark.createDataFrame(data.map(Tuple1.apply)).toDF("features")
    test.show
    //    +--------------+
    //    |      features|
    //    +--------------+
    //    |[-1.0,1.5,1.3]|
    //    |[3.0,2.0,-0.1]|
    //    |[0.0,2.2,-1.5]|
    //    +--------------+

    rModel.transform(test).show(false)
    //    +--------------+----------+
    //    |features      |prediction|
    //    +--------------+----------+
    //    |[-1.0,1.5,1.3]|1.0       |
    //    |[3.0,2.0,-0.1]|0.0       |
    //    |[0.0,2.2,-1.5]|1.0       |
    //    +--------------+----------+