隨機森林

做分類
做迴歸，即預測

多個決策樹構成，通過多個決策樹投票結果分數進行分類，不容易出現過度擬合
在生成的過程當中分別在行方向和 列方向上新增隨機過程，行方向上構建決策樹時採用放回抽樣（bootstraping）得到訓練資料，列方向上採用無放 回隨機抽樣得到特徵子集，並據此得到其最優切分點

Spark ML優化隨機森林

在 Spark 平臺上，傳統 單機形式的迭代方式必須要進行相應改進才能適用於分散式環境，這是因為在分散式環境下，資料也 是分散式的，演算法設計不得當會生成大量的IO 操作，影響演算法效率
三個優化策略
1.切分點抽樣統計
在單機環境下的決策樹對連續變數進行切分點選擇時，一般是通過對特徵點進 行排序，然後取相鄰兩個數之間的點作為切分點，
如果在分散式環境下 如此操作的話，會帶來大量的網路傳輸操作，特別是當資料量達到 PB 級時，演算法效率將極為低下
Spark 中的隨機森林在構建決策樹時，會對各分割槽採用一定的子特徵策略進行抽樣， 然後生成各個分割槽的統計資料，並最終得到切分點
2.切分特徵裝箱（Binning）
決策樹的構建過程就是對特徵的取值不斷進行劃分的過程
對於離散 的特徵，如果有 M 個值，最多 個劃分
如果值是有序的，那麼就最多 M-1 個劃分（按老，中，少的序，那麼只有 m-1 個，即 2 種劃分，老|中，少；老，中|少）
劃分的點就是 split（切分點），劃分出的區間就是 bin。對於連續特徵 ，理論上 split 是無數的，在分佈環境下不可能取出所有的值，因此它採用的是1中的切點抽樣統計方 法
3.逐層訓練（level-wise training）
單機版本的決策數生成過程是通過遞迴呼叫（本 質上是深度優先）的方式構造樹，在構造樹的同時，需要移動資料，將同一個子節點的資料移動到一 起
分散式環境下采用的策略是逐層構建樹節點（本質上是廣度優先），這樣遍歷所有資料的次數 等於所有樹中的最大層數。每次遍歷時，只需要計算每個節點所有切分點統計引數，遍歷完後，根據 節點的特徵劃分，決定是否切分，以及如何切分

使用Spark ML分別進行迴歸與分類建模

0 原資料集的模樣

No,year,month,day,hour,pm,DEWP,TEMP,PRES,cbwd,Iws,Is,Ir
1,2010,1,1,0,NaN,-21.0,-11.0,1021.0,NW,1.79,0.0,0.0
2,2010,1,1,1,NaN,-21,-12,1020,NW,4.92,0,0
3,2010,1,1,2,NaN,-21,-11,1019,NW,6.71,0,0
4,2010,1,1,3,NaN,-21,-14,1019,NW,9.84,0,0
5,2010,1,1,4,NaN,-20,-12,1018,NW,12.97,0,0
6,2010,1,1,5,NaN,-19
 
,-10,1017,NW,16.1,0,0
7,2010,1,1,6,NaN,-19,-9,1017,NW,19.23,0,0
8,2010,1,1,7,NaN,-19,-9,1017,NW,21.02,0,0
9,2010,1,1,8,NaN,-19,-9,1017,NW,24.15,0,0
10,2010,1,1,9,NaN,-20,-8,1017,NW,27.28,0,0
11,2010,1,1,10,NaN,-19,-7,1017,NW,31.3,0,0
12,2010,1,1,11,NaN,-18,-5,1017,NW,34.43,0,0
13,2010,1,1,12,NaN,-19,-5,1015,NW,37.56,0,0
14,2010,1,1,13,NaN,-18,-3
 
,1015,NW,40.69,0,0
15,2010,1,1,14,NaN,-18,-2,1014,NW,43.82,0,0
16,2010,1,1,15,NaN,-18,-1,1014,cv,0.89,0,0
……

1 讀取pm.csv,將含有缺失值的行扔掉（或用均值填充）將資料集分為兩部分，0.8比例作為訓練集，0.2比例作為測試集

case class data(No: Int, year: Int, month: Int, day: Int, hour: Int, pm: Double, DEWP: Double, TEMP: Double, PRES: Double, cbwd: String, Iws: Double, Is: Double, Ir: Double)

  def main(args: Array[String]): Unit = {
    val conf = new SparkConf().setMaster("local[*]").setAppName("foreast")
    val sc = new SparkContext(conf)
    val sqlContext = new SQLContext(sc)
    Logger.getLogger("org").setLevel(Level.ERROR)

    val root = MyRandomForeast.getClass.getResource("/")
    import sqlContext.implicits._

    val df = sc.textFile(root + "pm.csv").map(_.split(",")).filter(_ (0) != "No").filter(!_.contains("NaN")).map(x => data(x(0).toInt, x(1).toInt, x(2).toInt, x(3).toInt, x(4).toInt, x(5).toDouble, x(6).toDouble, x(7).toDouble, x(8).toDouble, x(9), x(10).toDouble, x(11).toDouble, x(12).toDouble)).toDF.drop("No").drop("year")

2 使用month,day,hour,DEWP,TEMP,PRES,cbwd,Iws,Is,Ir作為特徵列（除去No，year，
pm），pm作為label列，使用訓練集、隨機森林演算法進行迴歸建模，使用迴歸模型對測試集進行預測，並評估。

val splitdf = df.randomSplit(Array(0.8, 0.2))
    val (train, test) = (splitdf(0), splitdf(1))
    val traindf = train.withColumnRenamed("pm", "label")

    val indexer = new StringIndexer().setInputCol("cbwd").setOutputCol("cbwd_")
    val assembler = new VectorAssembler().setInputCols(Array("month", "day", "hour", "DEWP", "TEMP", "PRES", "cbwd_", "Iws", "Is", "Ir")).setOutputCol("features")

    import org.apache.spark.ml.Pipeline

    val rf = new RandomForestRegressor().setLabelCol("label").setFeaturesCol("features")
   //    setMaxDepth最大20，會大幅增加計算時間，增大能有效減小根均方誤差
    //setMaxBins我覺得和資料量相關，單獨增大適得其反，要和setNumTrees一起增大
    //目前這個引數得到的評估結果（根均方誤差）46.0002637676162
    //    val numClasses=2
    //    val categoricalFeaturesInfo = Map[Int, Int]()// Empty categoricalFeaturesInfo indicates all features are continuous.

    val pipeline = new Pipeline().setStages(Array(indexer, assembler, rf))
    val model = pipeline.fit(traindf)

    val testdf = test.withColumnRenamed("pm", "label")
    val labelsAndPredictions = model.transform(testdf)

    labelsAndPredictions.select("prediction", "label", "features").show(false)

    val eva = new RegressionEvaluator().setLabelCol("label").setPredictionCol("prediction")
    val accuracy = eva.evaluate(labelsAndPredictions)
    println(accuracy)

    val treemodel = model.stages(2).asInstanceOf[RandomForestRegressionModel]
    println("Learned regression forest model:\n" + treemodel.toDebugString)

3 按照下面標準處理pm列，數字結果放進（levelNum）列，字串結果放進（levelStr）列
優（0） 50
良（1）50~100
輕度汙染（2） 100~150
中度汙染（3） 150~200
重度汙染（4） 200~300
嚴重汙染（5） 大於300及以上

//使用Bucketizer特徵轉換，按區間劃分
val splits = Array(Double.NegativeInfinity, 50, 100, 150, 200, 300, Double.PositiveInfinity)

    val bucketizer = new Bucketizer().setInputCol("pm").setOutputCol("levelNum").setSplits(splits)

    val bucketizerdf = bucketizer.transform(df)

    val tempdf = sqlContext.createDataFrame(
      Seq((0.0, "優"), (1.0, "良"), (2.0, "輕度汙染"), (3.0, "中度汙染"), (4.0, "重度汙染"), (5.0, "嚴重汙染"))
    ).toDF("levelNum", "levelStr")
    val df2 = bucketizerdf.join(tempdf, "levelNum").drop("pm")

    df2 show

4 使用month,day,hour,DEWP,TEMP,PRES,cbwd,Iws,Is,Ir作為特徵列（除去No，year，pm）,levelNum作為label列，使用訓練集、隨機森林演算法進行分類建模。使用分類模型對測試集進行預測對預測結果df進行處理，基於prediction列生成predictionStr（0-5轉換優-嚴重汙染）,對結果進行評估

val splitdf2 = df2.randomSplit(Array(0.8, 0.2))
    val (train2, test2) = (splitdf2(0), splitdf2(1))
    val traindf2 = train2.withColumnRenamed("levelNum", "label")

    val indexer2 = new StringIndexer().setInputCol("cbwd").setOutputCol("cbwd_")

    val assembler2 = new VectorAssembler().setInputCols(Array("month", "day", "hour", "DEWP", "TEMP", "PRES", "cbwd_", "Iws", "Is", "Ir")).setOutputCol("features")

    val rf2 = new RandomForestClassifier().setLabelCol("label").setFeaturesCol("features")

    val pipeline2 = new Pipeline().setStages(Array(indexer2, assembler2, rf2))
    val model2 = pipeline2.fit(traindf2)

    val testdf2 = test2.withColumnRenamed("levelNum", "label")
    val labelsAndPredictions2 = model2.transform(testdf2)

    labelsAndPredictions2.select("label", "prediction", "features").show

    val eva2 = new MulticlassClassificationEvaluator().setLabelCol("label").setPredictionCol("prediction").setMetricName("precision")
    val accuracy2 = eva2.evaluate(labelsAndPredictions2)
    println("Test Error = " + (1.0 - accuracy2))

    val treeModel2 = model2.stages(2).asInstanceOf[RandomForestClassificationModel]
    println("learned classifier tree node:\n" + treeModel2.toDebugString)