Spark 實踐——用決策樹算法預測森林植被

本文基於《Spark 高級數據分析》第4章 用決策樹算法預測森林植被集。
完整代碼見 https://github.com/libaoquan95/aasPractice/tree/master/c4/rdf

1.獲取數據集

本 章 用 到 的 數 據 集 是 著 名 的 Covtype 數 據 集， 該 數 據 集 可 以 在 線 下 載（http://t.cn/R2wmIsI），包含一個 CSV 格式的壓縮數據文件 covtype.data.gz，附帶一個描述數據文件的信息文件 covtype.info。

該數據集記錄了美國科羅拉多州不一樣地塊的森林植被類型（也就是現實中的森林，這僅僅是巧合！）每一個樣本包含了描述每塊土地的若干特徵，包括海拔、坡度、到水源的距離、遮陽狀況和土壤類型， 而且隨同給出了地塊的已知森林植被類型。咱們須要總共 54 個特徵中的其他各項來預測森林植被類型。

人們已經用該數據集進行了研究，甚至在 Kaggle 大賽（https://www.kaggle.com/c/forestcover-type-prediction） 中也用過它。本章之因此研究這個數據集， 緣由在於它不但包含了數值型特徵並且包含了類別型特徵。 該數據集有 581 012 個樣本，雖然還稱不上大數據，但做爲一個範例來已經足夠大，並且也可以反映出大數據上的一些問題。

下載地址：

1. http://t.cn/R2wmIsIl （原書地址）
2. https://github.com/libaoquan95/aasPractice/tree/master/c4/covtype

2.數據處理

加載數據

val dataDir = "covtype.data"
val dataWithoutHeader = sc.read. option("inferSchema", true).option("header", false). csv(dataDir)
dataWithoutHeader.printSchema

結構化數據

val colNames = Seq(
    "Elevation", "Aspect", "Slope",
    "Horizontal_Distance_To_Hydrology", "Vertical_Distance_To_Hydrology",
    "Horizontal_Distance_To_Roadways",
    "Hillshade_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"))

val Array(trainData, testData) = data.randomSplit(Array(0.9, 0.1))
trainData.cache()
testData.cache()

data.printSchema

3.構造決策樹

構造特徵向量

val inputCols = trainData.columns.filter(_ != "Cover_Type")
val assembler = new VectorAssembler().setInputCols(inputCols).setOutputCol("featureVector")
val assembledTrainData = assembler.transform(trainData)

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 f1 = evaluator.setMetricName("f1").evaluate(predictions)
println(accuracy)
println(f1)

val predictionRDD = predictions.
    select("prediction", "Cover_Type").
    as[(Double,Double)].rdd
val multiclassMetrics = new MulticlassMetrics(predictionRDD)
println(multiclassMetrics.confusionMatrix)

val confusionMatrix = predictions.
    groupBy("Cover_Type").
    pivot("prediction", (1 to 7)).
    count().
    na.fill(0.0).
    orderBy("Cover_Type")

confusionMatrix.show()