Spark MLib 每季一算法 —— DecisionTree

DecisionTree(决策树)

DecisionTree通常被用来做分类器，但算法本身也可以做回归(Regression)模型的。相比逻辑回归(Logistic regression)和线性回归(Linear regression)，使用决策树建立的分类器有很好的可解释新(人类能知道为什么)，能够直接处理离散型的属性(feature)数据，不需要对所有属性值进行归一化处理，能更好的识别非线性对应关系。而缺点就是过拟合比较严重，需要通过剪枝来解决，不过现在有ensemble method可以来处理，所以Spark里不提供对单个树剪枝的方法以及参数。

Spark中根据决策树的用法，实现了2个Estimator类，分别是用作分类(处理label是类别)的DecisionTreeClassifier和用作回归(处理label是连续数值)的DecisionTreeRegressor，从而两者也在fit训练数据后产出不同的模型对象，DecisionTreeClassificationModel和DecisionTreeRegressionModel。两者其实本质上没有差别，核心的不同就是IG(information gain)算法不一样，Spark之所以分2个类，主要是出于工程实现的原因，因为在Spark里回归和分类有一些统一处理的逻辑在各自的基类继承树中需要被重用。数据集被模型对象transform后的输出都包含了最终预测的分类或数值(prediction)，另外DecisionTreeClassificationModel额外输出测试数据最终所在叶子节点的各个label的个数(rawPrediction)以及这些个数占总数的比值(probability)，而DecisionTreeRegressionModel额外输出测试数据最终所在叶子节点的所有label值的方差(variance)

伪算法过程

树的生成和使用
决策树算法的核心过程就是将整个训练集做为整颗树的根节点，然后根据是否能够使得整个树的叶子结点纯度和(node impurity)变得更高来决定当前节点是否需要分裂成子节点，因为每次分解后不可逆，所以这样生成的树未必是全局最优解(即所有叶子节点的纯度和在所有可能的树结构下最高)。
为了计算纯度需要引入了信息增益(information gain)的概念，在数据集D上的某个划分方式s的信息增益记作IG(D,s)，所以分裂过程就是寻找一个按属性的划分方式使得IG(D,s)最大的过程。最终通过递归这一分裂过程得到一颗完整的树，其中每个叶子节点包含的划分后剩下的数据集的label(实际的树可能会出现多个分类值或者是多个连续数据值的情况)表示最终的预测分类和数值结果；新的测试数据会从这颗树的根节点开始，每通过一个非叶节点就会根据实际的属性值走向不同的路径，最终到达叶子节点从而得到预测结果。

信息增益函数(IG)
Spark中使用二划分的方式来分裂节点，所以一个节点要么不分裂，如果分裂就会分成左右2个子节点。分裂过程中的信息增益函数公式如下：
$$IG(D,s) = Impurity(D) - \frac{N_{left}}{N}Impurity(D_{left}) - \frac{N_{right}}{N}Impurity(D_{right})$$
其中D为节点数据集，N为节点数据集包含的记录数，Impurity()函数得到节点纯度，该公式表示——信息增益为当前待分裂节点的纯度值减去分裂后左右节点的纯度值按记录数的加权和。纯度计算公式在Spark中可以自己选择，总共有3种算法，但其实就是1/2里面二选一：

1. 基尼指数(Gini impurity)，只用于分类算法：$\sum_{i=1}^{C}f_{i}(1-f_{i})$
2. 信息熵(Entropy)，只用于分类算法：$\sum_{i=1}^{C}-f_{i}log(f_{i})$
3. 方差(Variance)，只用于回归算法：$\frac{1}{N}\sum_{i=1}^{N}(y_{i}-μ)^{2}$

其中$f_{i}$是节点数据集中某个分类label出现的频率。

节点分裂时的划分规则
当节点分裂时，在计算IG之前还需要确定划分方式，这是因为现实中在还有大量属性值的情况下，无法逐个计算所有划分方式带来的IG值。
对于属性值连续的情况，Spark采用分箱的方式来做数据集划分，用户可以指定最大分箱个数(maxBins)，箱体数量确定后，Spark使用近似分位数算法(和之前文章介绍过的QuantileDiscretizer中使用的算法一样)将连续数据处理成离散数据。
对于离散类别属性值，所有可能的备选划分方案总共有$2^{M−1}−1$，其中M是类别属性值的数量。由此可见，当有些属性如果属性值非常多的情况下，IG计算量是非常大的。在做回归和二分类(label非0即1)决策树的情况下，备选划分可以被简化，这时Spark先对数据集中具有某一个属性值的数据的label值做一次求均值，然后根据均值排序，按序依次做出左右二分，所有这些划分合起来就是备选的划分方案，所以最终方案被缩减到M-1个。在多分类问题下，其实Spark在当$2^{M−1}−1$多于最大指定分箱数(maxBins)的情况下，也会做类似的简化，但需要预先计算每个类别数据的纯度值，然后按照纯度值排序依次做划分，相当于也缩减到M-1个划分。

参数

输入输出相关:

• labelCol: 输入训练集中标记字段的名称 (默认值: label)
• featuresCol: 输入训练集和预测数据集中的属性向量的字段名称(默认值: features)
• DecisionTreeClassifier.rawPredictionCol: 输出结果数据中存放命中的叶子节点里各个label个数的字段名称 (默认值: rawPrediction)
• DecisionTreeClassifier.probabilityCol: 输出结果数据中存放命中的叶子节点里各个label个数占比的字段名称 (默认值: probability)
• DecisionTreeClassifier.predictionCol: 输出结果数据中最终判别类别的字段名称 (默认值: prediction)
• DecisionTreeRegressor.predictionCol: 输出结果数据中最终预测的数据值的字段名称 (默认值: prediction)
• DecisionTreeRegressor.varianceCol: 输出结果数据中存放命中的叶子节点的label值的标准差的字段名称 (无默认值，默认不输出)

算法模型相关:

• DecisionTreeClassifier.impurity: 纯度计算公式(默认值：gini)
  可选项1. gini-基尼指数
  可选项2. entropy-信息熵
• DecisionTreeRegressor.impurity: 纯度计算公式(默认值：variance)
  可选项1. variance-方差值，回归情况下暂时就支持这一种计算方式。
• maxBins: 最大分箱数，在对连续属性分箱和离散属性做划分时用到，具体使用方式见上面划分规则部分相关介绍，该值必须不少于任何一个离散属性的类别数。(>=2，默认值: 32)
• maxDepth: 树的最大深度，达到最大深度后，节点不再分裂。(>=0，0代表就只有根节点，默认值: 5)
• minInfoGain: 分裂时需要达到的最小IG数值，如果达不到这个值，则节点不再分裂，默认只要有IG就会分裂。(>=0.0实数，默认值: 0.0)
• minInstancesPerNode: 树节点包含的最小数据个数，如果分裂时左右出现任意一个节点的数据量少于该值，则节点不分裂。(>=1，默认值: 1)
• DecisionTreeClassifier.thresholds: 多分类算法中因为树的叶子节点不一定是含有纯label的数据集，所以最终给出的是某个label的频率值，用户可以通过设置这个数组参数，将频率值映射到最终分类类别上，该参数数组长度必须和类别数量吻合，最终选择的分类是频率值除以该分类的threshold值最大的那个，如果没有设置，相当于直接取频率最大的。(无默认值)

model对象的成员

• depth: 决策树最终的高度，从0开始，0代表只有根节点。
• numNodes: 决策树包含的节点总数，包括内部节点和叶子节点。
• rootNode: 决策树的根节点对象(类型是org.apache.spark.ml.tree.Node)，该对象里包含节点的纯度值impurity和节点的预测值prediction。
• featureImportances: 决策树的每一个属性值的重要程度(importance)组成的向量值，该值是所有以该属性分裂的节点的IG值的节点记录数加权和。
• 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 dtc = new DecisionTreeClassifier()

val cModel = dtc.fit(training)

println(s"DecisionTreeClassifier model's depth and numNodes: ${cModel.depth} / ${cModel.numNodes}")
//  1 / 3  (1个根节点，2个叶子节点)
println("DecisionTreeClassifier model's featureImportances: " + cModel.featureImportances)
//  [1.0, 0.0, 0.0] (按第一个feature就能划分出label)
println("DecisionTreeClassifier model's debug info: " + cModel.toDebugString)
//  If (feature 0 <= 1.0)
//    Predict: 1.0
//  Else (feature 0 > 1.0)
//    Predict: 0.0


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]|[0.0,2.0]    |[0.0,1.0]  |1.0       |
//    |[3.0,2.0,1.0]  |[2.0,0.0]    |[1.0,0.0]  |0.0       |
//    |[0.0,22.0,5.0] |[0.0,2.0]    |[0.0,1.0]  |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 dtr = new DecisionTreeRegressor().setVarianceCol("variance")

val rModel = dtr.fit(training)

println(s"DecisionTreeRegressor model's depth and numNodes: ${rModel.depth} / ${rModel.numNodes}")
//  2 / 5  (1个根节点，1个中间节点，3个叶子节点)
println("DecisionTreeRegressor model's featureImportances: " + rModel.featureImportances)
//  [0.6666666666666666, 0.3333333333333333, 0.0] (按前2个feature就能预测出label)
println("DecisionTreeRegressor model's debug info: " + rModel.toDebugString)
//  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


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|variance|
//    +--------------+----------+--------+
//    |[-1.0,1.5,1.3]|1.0       |0.0     |
//    |[3.0,2.0,-0.1]|0.0       |0.0     |
//    |[0.0,2.2,-1.5]|1.0       |0.0     |
//    +--------------+----------+--------+
//    叶子节点都只有一条记录，所以这些节点的label数据的方差都是0