決策樹分類器

一.方法簡介

決策樹是一種基本的分類與回歸方法，這里主要介紹用於分類的決策樹。決策樹模式呈樹形結構，其中每

個內部節點表示一個屬性上的測試，每個分支代表一個測試輸出，每個葉節點代表一個類別。學習時利用

訓練數據，根據損失函數最小化的原則建立決策樹模型；預測時，對新的數據，利用決策樹模型進行分類

二.基本原理

決策樹學習通常包括三個步驟:特征選擇，決策樹的生成和決策樹的剪枝

(1)特征選擇

特征選擇在於選取對訓練數據具有分類能力的特征，這樣可以提高決策樹學習的效率。通常特征選擇的准則是信息增益（或信息增益比、

基尼指數等），每次計算每個特征的信息增益，並比較它們的大小，選擇信息增益最大（信息增益比最大、基尼指數最小）的特征。下面

我們重點介紹一下特征選擇的准則：信息增益。首先定義信息論中廣泛使用的一個度量標准——熵（entropy），它是表示隨機變量不確

定性的度量。熵越大，隨機變量的不確定性就越大。而信息增益（informational entropy）表示得知某一特征后使得信息的不確定性減

少的程度。簡單的說，一個屬性的信息增益就是由於使用這個屬性分割樣例而導致的期望熵降低。信息增益、信息增益比和基尼指數的具

體定義如下：信息增益：特征A對訓練數據集D的信息增益，定義為集合D的經驗熵與特征A給定條件下D的經驗條件熵之差，

即信息增益比：特征A對訓練數據集D的信息增益比定義為其信息增益與訓練數據集D關於特征A的值的熵之比，即其中，，n是特征A取值的個

數。基尼指數：分類問題中，假設有K個類，樣本點屬於第K類的概率，則概率分布的基尼指數定義為 

（二）決策樹的生成

​ 從根結點開始，對結點計算所有可能的特征的信息增益，選擇信息增益最大的特征作為結點的特征，由該特征的不同取值建立子結點，再對

子結點遞歸地調用以上方法，構建決策樹；直到所有特征的信息增均很小或沒有特征可以選擇為止，最后得到一個決策樹。

​ 決策樹需要有停止條件來終止其生長的過程。一般來說最低的條件是：當該節點下面的所有記錄都屬於同一類，或者當所有的記錄屬性都具

有相同的值時。這兩種條件是停止決策樹的必要條件，也是最低的條件。在實際運用中一般希望決策樹提前停止生長，限定葉節點包含的最低

數據量，以防止由於過度生長造成的過擬合問題。

（三）決策樹的剪枝

​ 決策樹生成算法遞歸地產生決策樹，直到不能繼續下去為止。這樣產生的樹往往對訓練數據的分類很准確，但對未知的測試數據的分類卻沒有

那么准確，即出現過擬合現象。解決這個問題的辦法是考慮決策樹的復雜度，對已生成的決策樹進行簡化，這個過程稱為剪枝。

​ 決策樹的剪枝往往通過極小化決策樹整體的損失函數來實現。一般來說，損失函數可以進行如下的定義： 其中，

T為任意子樹， 為對訓練數據的預測誤差（如基尼指數），為子樹的葉結點個數，為參數，為參數是時的子樹T的整

體損失，參數權衡訓練數據的擬合程度與模型的復雜度。對於固定的，一定存在使損失函數最小的子樹，將其表示為。當大的時

候，最優子樹偏小；當小的時候，最優子樹 偏大。

 

#導入需要的包

from pyspark import SparkContext

from pyspark.sql import SparkSession,Row

from pyspark.ml.feature import IndexToString, StringIndexer, VectorIndexer

from pyspark.ml import Pipeline

from pyspark.ml.linalg import Vector,Vectors

from pyspark.ml.evaluation import MulticlassClassificationEvaluator

from pyspark.ml.classification import DecisionTreeClassificationModel,DecisionTreeClassifier

from pyspark.ml.regression import DecisionTreeRegressionModel,DecisionTreeRegressor

from pyspark.ml.evaluation import RegressionEvaluator

 

sc = SparkContext('local','決策樹分類器')

spark = SparkSession.builder.master('local').appName('決策樹分類器').getOrCreate()

#讀取數據，簡要分析

def f(x):

rel = {}

rel['features'] = Vectors.dense(float(x[0]), float(x[1]), float(x[2]), float(x[3]))

rel['label'] = str(x[4])

return rel

data = sc.textFile("file:///usr/local/spark/mycode/exercise/iris.txt").map(lambda line: line.split(',')).map(lambda p: Row(**f(p))).toDF()

data.createOrReplaceTempView("iris")

df =spark.sql("select * from iris")

rel = df.rdd.map(lambda t: str(t[1])+":"+str(t[0])).collect()

# for item in rel:

# print(item)

#進一步處理特征和標簽，以及數據分組

#1.分別獲取標簽列和特征列，進行索引，並進行了重命名

labelIndexer = StringIndexer().setInputCol("label").setOutputCol("indexedLabel").fit(df)

featuresIndexer = VectorIndexer().setInputCol("features").setOutputCol("indexedFeatures").fit(df)

#這里我們設置一個labelConverter，目的是把預測的類別重新轉化成字符型的

labelConverter = IndexToString().setInputCol("prediction").setOutputCol("predictedLabel").setLabels(labelIndexer.labels)

#把數據分為訓練集與測試集

trainingData, testData = data.randomSplit([0.7,0.3])

#訓練決策樹模型

dt = DecisionTreeClassifier().setLabelCol("indexedLabel").setFeaturesCol("indexedFeatures")

#在Pipeline中進行設置

pipelineClassfier = Pipeline().setStages([labelIndexer, featuresIndexer, dt, labelConverter])

#訓練決策樹模型

modelClassifer = pipelineClassfier.fit(trainingData)

#進行預測

predictionClassifier = modelClassifer.transform(testData)

#查看部分預測結果

predictionClassifier.select("predictedLabel", "label", "features").show(20)

 

#2.評估決策樹模型

evaluatorClassifier = MulticlassClassificationEvaluator().setLabelCol("indexedLabel").setPredictionCol("prediction").setMetricName("accuracy")

accuracy = evaluatorClassifier.evaluate(predictionClassifier)

print("Descion Tree Test Error:"+str(1.0-accuracy))

 

treeModelClassifier = modelClassifer.stages[2]

print("Learned classification tree model:\n"+str(treeModelClassifier.toDebugString))

 

#3.構建決策樹回歸模型

#訓練決策樹模型

dtRegressor = DecisionTreeRegressor().setLabelCol("indexedLabel").setFeaturesCol("indexedFeatures")

#在pipeline中進行訓練

pipelineReggressor =Pipeline().setStages([labelIndexer, featuresIndexer, dtRegressor, labelConverter])

#訓練決策樹模型

modelRegressor = pipelineReggressor.fit(trainingData)

#進行測試

predictionRegressor = modelRegressor.transform(testData)

#查看部分結果

predictionRegressor.select("predictedLabel", "label", "features").show(20)

 

#評估決策樹回歸模型

evalorRegressor = RegressionEvaluator().setLabelCol("indexedLabel").setPredictionCol("prediction").setMetricName("rmse")

rmse = evalorRegressor.evaluate(predictionRegressor)

print("Root Mean Squared Error (RMSE) on test data= "+ str(rmse))

 

treeModelRegressor = modelRegressor.stages[2]

print("Learn regression tree model:\n"+ str(treeModelRegressor.toDebugString))