梯度迭代树（GBDT）算法原理及Spark MLlib调用实例（Scala/Java/python）

http://blog.csdn.net/liulingyuan6/article/details/53426350

梯度迭代树

算法简介：

梯度提升树是一种决策树的集成算法。它通过反复迭代训练决策树来最小化损失函数。决策树类似，梯度提升树具有可处理类别特征、易扩展到多分类问题、不需特征缩放等性质。Spark.ml通过使用现有decision tree工具来实现。

梯度提升树依次迭代训练一系列的决策树。在一次迭代中，算法使用现有的集成来对每个训练实例的类别进行预测，然后将预测结果与真实的标签值进行比较。通过重新标记，来赋予预测结果不好的实例更高的权重。所以，在下次迭代中，决策树会对先前的错误进行修正。

对实例标签进行重新标记的机制由损失函数来指定。每次迭代过程中，梯度迭代树在训练数据上进一步减少损失函数的值。spark.ml为分类问题提供一种损失函数（Log Loss），为回归问题提供两种损失函数（平方误差与绝对误差）。

Spark.ml支持二分类以及回归的随机森林算法，适用于连续特征以及类别特征。

＊注意梯度提升树目前不支持多分类问题。

参数：

checkpointInterval:

类型：整数型。

含义：设置检查点间隔（>=1），或不设置检查点（-1）。

featuresCol:

类型：字符串型。

含义：特征列名。

impurity:

类型：字符串型。

含义：计算信息增益的准则（不区分大小写）。

labelCol:

类型：字符串型。

含义：标签列名。

lossType:

类型：字符串型。

含义：损失函数类型。

maxBins:

类型：整数型。

含义：连续特征离散化的最大数量，以及选择每个节点分裂特征的方式。

maxDepth:

类型：整数型。

含义：树的最大深度（>=0）。

maxIter:

类型：整数型。

含义：迭代次数（>=0）。

minInfoGain:

类型：双精度型。

含义：分裂节点时所需最小信息增益。

minInstancesPerNode:

类型：整数型。

含义：分裂后自节点最少包含的实例数量。

predictionCol:

类型：字符串型。

含义：预测结果列名。

rawPredictionCol:

类型：字符串型。

含义：原始预测。

seed:

类型：长整型。

含义：随机种子。

subsamplingRate:

类型：双精度型。

含义：学习一棵决策树使用的训练数据比例，范围[0,1]。

stepSize:

类型：双精度型。

含义：每次迭代优化步长。

示例：

下面的例子导入LibSVM格式数据，并将之划分为训练数据和测试数据。使用第一部分数据进行训练，剩下数据来测试。训练之前我们使用了两种数据预处理方法来对特征进行转换，并且添加了元数据到DataFrame。

Scala:

1. import org.apache.spark.ml.Pipeline
2. import org.apache.spark.ml.classification.{GBTClassificationModel, GBTClassifier}
3. import org.apache.spark.ml.evaluation.MulticlassClassificationEvaluator
4. import org.apache.spark.ml.feature.{IndexToString, StringIndexer, VectorIndexer}
5. // Load and parse the data file, converting it to a DataFrame.
6. val data = spark.read.format(“libsvm”).load(“data/mllib/sample_libsvm_data.txt”)
7. // Index labels, adding metadata to the label column.
8. // Fit on whole dataset to include all labels in index.
9. val labelIndexer = new StringIndexer()
10. .setInputCol(“label”)
11. .setOutputCol(“indexedLabel”)
12. .fit(data)
13. // Automatically identify categorical features, and index them.
14. // Set maxCategories so features with > 4 distinct values are treated as continuous.
15. val featureIndexer = new VectorIndexer()
16. .setInputCol(“features”)
17. .setOutputCol(“indexedFeatures”)
18. .setMaxCategories(4)
19. .fit(data)
20. // Split the data into training and test sets (30% held out for testing).
21. val Array(trainingData, testData) = data.randomSplit(Array(0.7, 0.3))
22. // Train a GBT model.
23. val gbt = new GBTClassifier()
24. .setLabelCol(“indexedLabel”)
25. .setFeaturesCol(“indexedFeatures”)
26. .setMaxIter(10)
27. // Convert indexed labels back to original labels.
28. val labelConverter = new IndexToString()
29. .setInputCol(“prediction”)
30. .setOutputCol(“predictedLabel”)
31. .setLabels(labelIndexer.labels)
32. // Chain indexers and GBT in a Pipeline.
33. val pipeline = new Pipeline()
34. .setStages(Array(labelIndexer, featureIndexer, gbt, labelConverter))
35. // Train model. This also runs the indexers.
36. val model = pipeline.fit(trainingData)
37. // Make predictions.
38. val predictions = model.transform(testData)
39. // Select example rows to display.
40. predictions.select(“predictedLabel”, “label”, “features”).show(5)
41. // Select (prediction, true label) and compute test error.
42. val evaluator = new MulticlassClassificationEvaluator()
43. .setLabelCol(“indexedLabel”)
44. .setPredictionCol(“prediction”)
45. .setMetricName(“accuracy”)
46. val accuracy = evaluator.evaluate(predictions)
47. println(“Test Error = ” + (1.0 – accuracy))
48. val gbtModel = model.stages(2).asInstanceOf[GBTClassificationModel]
49. println(“Learned classification GBT model:\n” + gbtModel.toDebugString)

Java:

1. import org.apache.spark.ml.Pipeline;
2. import org.apache.spark.ml.PipelineModel;
3. import org.apache.spark.ml.PipelineStage;
4. import org.apache.spark.ml.classification.GBTClassificationModel;
5. import org.apache.spark.ml.classification.GBTClassifier;
6. import org.apache.spark.ml.evaluation.MulticlassClassificationEvaluator;
7. import org.apache.spark.ml.feature.*;
8. import org.apache.spark.sql.Dataset;
9. import org.apache.spark.sql.Row;
10. import org.apache.spark.sql.SparkSession;
11. // Load and parse the data file, converting it to a DataFrame.
12. Dataset<Row> data = spark
13. .read()
14. .format(“libsvm”)
15. .load(“data/mllib/sample_libsvm_data.txt”);
16. // Index labels, adding metadata to the label column.
17. // Fit on whole dataset to include all labels in index.
18. StringIndexerModel labelIndexer = new StringIndexer()
19. .setInputCol(“label”)
20. .setOutputCol(“indexedLabel”)
21. .fit(data);
22. // Automatically identify categorical features, and index them.
23. // Set maxCategories so features with > 4 distinct values are treated as continuous.
24. VectorIndexerModel featureIndexer = new VectorIndexer()
25. .setInputCol(“features”)
26. .setOutputCol(“indexedFeatures”)
27. .setMaxCategories(4)
28. .fit(data);
29. // Split the data into training and test sets (30% held out for testing)
30. Dataset<Row>[] splits = data.randomSplit(new double[] {0.7, 0.3});
31. Dataset<Row> trainingData = splits[0];
32. Dataset<Row> testData = splits[1];
33. // Train a GBT model.
34. GBTClassifier gbt = new GBTClassifier()
35. .setLabelCol(“indexedLabel”)
36. .setFeaturesCol(“indexedFeatures”)
37. .setMaxIter(10);
38. // Convert indexed labels back to original labels.
39. IndexToString labelConverter = new IndexToString()
40. .setInputCol(“prediction”)
41. .setOutputCol(“predictedLabel”)
42. .setLabels(labelIndexer.labels());
43. // Chain indexers and GBT in a Pipeline.
44. Pipeline pipeline = new Pipeline()
45. .setStages(new PipelineStage[] {labelIndexer, featureIndexer, gbt, labelConverter});
46. // Train model. This also runs the indexers.
47. PipelineModel model = pipeline.fit(trainingData);
48. // Make predictions.
49. Dataset<Row> predictions = model.transform(testData);
50. // Select example rows to display.
51. predictions.select(“predictedLabel”, “label”, “features”).show(5);
52. // Select (prediction, true label) and compute test error.
53. MulticlassClassificationEvaluator evaluator = new MulticlassClassificationEvaluator()
54. .setLabelCol(“indexedLabel”)
55. .setPredictionCol(“prediction”)
56. .setMetricName(“accuracy”);
57. double accuracy = evaluator.evaluate(predictions);
58. System.out.println(“Test Error = ” + (1.0 – accuracy));
59. GBTClassificationModel gbtModel = (GBTClassificationModel)(model.stages()[2]);
60. System.out.println(“Learned classification GBT model:\n” + gbtModel.toDebugString());

Python：

1. from pyspark.ml import Pipeline
2. from pyspark.ml.classification import GBTClassifier
3. from pyspark.ml.feature import StringIndexer, VectorIndexer
4. from pyspark.ml.evaluation import MulticlassClassificationEvaluator
5. # Load and parse the data file, converting it to a DataFrame.
6. data = spark.read.format(“libsvm”).load(“data/mllib/sample_libsvm_data.txt”)
7. # Index labels, adding metadata to the label column.
8. # Fit on whole dataset to include all labels in index.
9. labelIndexer = StringIndexer(inputCol=”label”, outputCol=”indexedLabel”).fit(data)
10. # Automatically identify categorical features, and index them.
11. # Set maxCategories so features with > 4 distinct values are treated as continuous.
12. featureIndexer =\
13. VectorIndexer(inputCol=”features”, outputCol=”indexedFeatures”, maxCategories=4).fit(data)
14. # Split the data into training and test sets (30% held out for testing)
15. (trainingData, testData) = data.randomSplit([0.7, 0.3])
16. # Train a GBT model.
17. gbt = GBTClassifier(labelCol=”indexedLabel”, featuresCol=”indexedFeatures”, maxIter=10)
18. # Chain indexers and GBT in a Pipeline
19. pipeline = Pipeline(stages=[labelIndexer, featureIndexer, gbt])
20. # Train model.  This also runs the indexers.
21. model = pipeline.fit(trainingData)
22. # Make predictions.
23. predictions = model.transform(testData)
24. # Select example rows to display.
25. predictions.select(“prediction”, “indexedLabel”, “features”).show(5)
26. # Select (prediction, true label) and compute test error
27. evaluator = MulticlassClassificationEvaluator(
28. labelCol=”indexedLabel”, predictionCol=”prediction”, metricName=”accuracy”)
29. accuracy = evaluator.evaluate(predictions)
30. print(“Test Error = %g” % (1.0 – accuracy))
31. gbtModel = model.stages[2]
32. print(gbtModel)  # summary only