【大数据分析】Ch.3：决策树预测森林覆盖

我们喝牛奶时的思考过程就是决策树的一种，通过不同的条件分支生成不断的结果，而结果又有新的条件分支生成新的结果……
上面的例子是：

为客人推荐宠物的机器人工作。它需要根据我们给定的资料来模拟，确定出决策树判断的优先条件，这里的条件有体重，腿数目，颜色三要素。

• 体重大于100kg与否，大于则不适合
  
• 若体重小于100Kg，是否是绿色，绿色则不适合
  

进而我们引出一个概念——Entropy（熵）。如上图所示，是两类数据类型的图形，当二者之一为0是，熵也为0,；当二者55开始，熵值最高，也就说复杂度最高。就反映了我们刚才说的，如果树的分支只偏向一个方向，那么其实树就是一个N节点的线段，复杂度为0，否则复杂度是nlongn。
而对于公式的计算，如下图所示：

需要注意的是在这个计算过程中，0log0视为0而不是负无穷。因为当某一个选项没出现时，它的出现率就是0而不是无穷。
分析以上式子的结构，对于outlook这一个因素，有三个分支，分别是overcast，sunny，rainy。然后统计第一个图中对应的yes/no个数，并根据其个数计算系数。如表格Outlook中对应的overcast出现了4个yes，0个no。其中，yes占据4/4也就是1，而no占据0/4也就是0。其他的也同理，                              （                      占                      据                      的                      份                      数                      /                      总                      数                      ）                      ∗                      l                      o                      g                      ∗                      （                      占                      据                      的                      份                      数                      /                      总                      数                      ）                          （占据的份数/总数）*log*（占据的份数/总数）               （占据的份数/总数）∗log∗（占据的份数/总数）就是对应的计算方式。

如果我们只想要计算出具体因素中的某一个具体的项，则需要来看下一个公式。

计算由分割产生的所有集合的加权平均值即可。

有了基本概念后，我们来看信息增益。

• 选择差异(信息增益)最大化的属性
  
• 最大化信息增益等于最小化平均熵
  
• 当属性A将集合S分割为子集Si时，我们计算平均熵。然后比较原始集合S的和与熵
  

• 它能够处理很高维度（feature很多）的数据，并且不用做特征选择
  原因：特征子集是随机选择的，子集大小可以控制。
  
• 在训练完后，它能够给出feature重要大小
  
• 训练速度快，容易做成并行化方法
  原因：训练时树与树之间是相互独立的
  
• 如果有很大一部分的特征遗失，仍可以维持准确度。
  缺陷：
  
• 随机森林已经被证明在某些噪音较大的分类或回归问题上会过拟合
  
• 对于有不同取值的属性的数据，取值划分较多的属性会对随机森林产生更大的影响，(跟决策树算法类似，以信息增益划分数据集的特征，都有这个问题，可以用信息增益比校正)所以随机森林在这种数据上产出的属性权值是不可信的。
  

import org.apache.spark.ml.{PipelineModel,Pipeline}
import org.apache.spark.ml.classification.{DecisionTreeClassifier,RandomForestClassifier,RandomForestClassificationModel}
import org.apache.spark.ml.evaluation.MulticlassClassificationEvaluator
import org.apache.spark.ml.feature.{VectorAssembler,VectorIndexer}
import org.apache.spark.ml.linalg.Vector
import org.apache.spark.ml.tuning.{ParamGridBuilder,TrainValidationSplit}
import org.apache.spark.mllib.evaluation.MulticlassMetrics
import org.apache.spark.sql.{DataFrame,SparkSession}
import org.apache.spark.sql.functions._
import scala.util.Random
val dataWithoutHeader = spark.read.option("inferSchema", true).option("header",false).csv("Data/cov/covtype.data" )
dataWithoutHeader.first
dataWithoutHeader.printSchema()
val colNames =Seq(
"Elevation", "Aspect", "Slope" ,
"Horizontal_Distance_To_Hydrology","Yertical_Distance_To_Hydrology" ,
"Horizontal_Distance_To_Roadways",
"HilIshade_9am","Hillshade_Noon","Hillshade_3pm",
"Horizontal_Distance_To_Fire_Points")++
((0 until 4 ).map(i => s"Wilderness_Area_$i"))++
((0 until 40).map(i => s"Soil_Type_$i"))++Seq("Cover_Type")
val data =dataWithoutHeader.toDF(colNames:_* ).withColumn("Cover_Type",$"Cover_Type".cast("double"))
data.printSchema()
val Array(trainData,testData) = data.randomSplit(Array(0.9,0.1))
trainData.cache( )
testData.cache( )
val inputCols = trainData.columns. filter(_!="Cover_Type")
val assembler = new VectorAssembler().setInputCols(inputCols).setOutputCol("featureVector")
val assembledTrainData = assembler.transform(trainData)
assembledTrainData.select("featureVector").show(truncate = false)
val classifier = new DecisionTreeClassifier().
setSeed(Random.nextLong( )).
setLabelCol("Cover_Type").
setFeaturesCol("featureVector").
setPredictionCol("prediction")
val model=classifier.fit(assembledTrainData)
println(model.toDebugString)
model.featureImportances.toArray.zip(inputCols).sorted.reverse.foreach(println)
val predictions = model.transform(assembledTrainData)
predictions.select( "Cover_Type" , "prediction" , "probability").show(truncate = false)
val evaluator = new MulticlassClassificationEvaluator().
setLabelCol("Cover_Type").
setPredictionCol("prediction")
val accuracy = evaluator.setMetricName("accuracy").evaluate(predictions)
val predictionRDD =predictions.
select( "prediction", "Cover_Type").
as[(Double,Double)].rdd
val multiclassMetrics = new MulticlassMetrics(predictionRDD)
println(multiclassMetrics.confusionMatrix)
val classifierHyper = new DecisionTreeClassifier().
setSeed(Random.nextLong()).
setLabelCol( "Cover_Type" ).
setFeaturesCol( "featureVector" ).
setPredictionCol ( "prediction" )
val pipeline = new Pipeline().setStages(Array(assembler,classifierHyper))
val paramGrid = new ParamGridBuilder()
.addGrid(classifier.impurity, Seq("gini", "entropy"))
.addGrid(classifier.maxDepth, Seq(1, 5))
.addGrid(classifier.maxBins, Seq(40, 80))
.addGrid(classifier.minInfoGain, Seq(0.0, 0.05))
.build()  
val multiclassEval = new MulticlassClassificationEvaluator().
setLabelCol("Cover_Type").
setPredictionCol("prediction").
setMetricName("accuracy")
val validator = new TrainValidationSplit().
setSeed(Random.nextLong()).
setEstimator(pipeline).
setEvaluator(multiclassEval).
setEstimatorParamMaps(paramGrid).
setTrainRatio(0.9)
val validatorModel=validator.fit(trainData)
val bestModel = validatorModel.bestModel
bestModel.asInstanceOf[PipelineModel].stages.last.extractParamMap
val paramsAndMetrics = validatorModel.validationMetrics.
zip(validatorModel.getEstimatorParamMaps).sortBy(-_._1)
paramsAndMetrics.foreach{case (metric,params)=>
println(metric)
println(params)
println()
}
validatorModel.validationMetrics.max
multiclassEval.evaluate(bestModel.transform(testData))