spark 決策樹分類算法demo

分類（Classification）

下面的例子說明了怎樣導入LIBSVM 數據文件，解析成RDD[LabeledPoint]，然后使用決策樹進行分類。GINI不純度作為不純度衡量標准並且樹的最大深度設置為5。最后計算了測試錯誤率從而評估算法的准確性。

from pyspark.mllib.regression import LabeledPoint from pyspark.mllib.tree import DecisionTree, DecisionTreeModel from pyspark.mllib.util import MLUtils # Load and parse the data file into an RDD of LabeledPoint. data = MLUtils.loadLibSVMFile(sc, 'data/mllib/sample_libsvm_data.txt') # Split the data into training and test sets (30% held out for testing) (trainingData, testData) = data.randomSplit([0.7, 0.3]) # Train a DecisionTree model. # Empty categoricalFeaturesInfo indicates all features are continuous. model = DecisionTree.trainClassifier(trainingData, numClasses=2, categoricalFeaturesInfo={}, impurity='gini', maxDepth=5, maxBins=32) # Evaluate model on test instances and compute test error predictions = model.predict(testData.map(lambda x: x.features)) labelsAndPredictions = testData.map(lambda lp: lp.label).zip(predictions) testErr = labelsAndPredictions.filter(lambda (v, p): v != p).count() / float(testData.count()) print('Test Error = ' + str(testErr)) print('Learned classification tree model:') print(model.toDebugString()) # Save and load model model.save(sc, "myModelPath") sameModel = DecisionTreeModel.load(sc, "myModelPath")

 

以下代碼展示了如何載入一個LIBSVM數據文件，解析成一個LabeledPointRDD，然后使用決策樹，使用Gini不純度作為不純度衡量指標，最大樹深度是5.測試誤差用來計算算法准確率。

1. # -*- coding:utf-8 -*-
2. """
3. 測試決策樹
4. """
6. import os
7. import sys
8. import logging
9. from pyspark.mllib.tree import DecisionTree,DecisionTreeModel
10. from pyspark.mllib.util import MLUtils
12. # Path for spark source folder
13. os.environ['SPARK_HOME']="D:\javaPackages\spark-1.6.0-bin-hadoop2.6"
14. # Append pyspark to Python Path
15. sys.path.append("D:\javaPackages\spark-1.6.0-bin-hadoop2.6\python")
16. sys.path.append("D:\javaPackages\spark-1.6.0-bin-hadoop2.6\python\lib\py4j-0.9-src.zip")
18. from pyspark import SparkContext
19. from pyspark import SparkConf
21. conf = SparkConf()
22. conf.set("YARN_CONF_DIR ", "D:\javaPackages\hadoop_conf_dir\yarn-conf")
23. conf.set("spark.driver.memory", "2g")
25. #conf.set("spark.executor.memory", "1g")
26. #conf.set("spark.python.worker.memory", "1g")
27. conf.setMaster("yarn-client")
28. conf.setAppName("TestDecisionTree")
29. logger = logging.getLogger('pyspark')
30. sc = SparkContext(conf=conf)
32. mylog = []
33. #載入和解析數據文件為 LabeledPoint RDDdata = MLUtils.loadLibSVMFile(sc,"/home/xiatao/machine_learing/")
34. #將數據拆分成訓練集合測試集
35. (trainingData,testData) = data.randomSplit([0.7,0.3])
37. ##訓練決策樹模型
38. #空的 categoricalFeauresInfo 代表了所有的特征都是連續的
39. model = DecisionTree.trainClassifier(trainingData, numClasses=2,categoricalFeaturesInfo={},impurity='gini',maxDepth=5,maxBins=32)
41. # 在測試實例上評估模型並計算測試誤差
43. predictions = model.predict(testData.map(lambda x:x.features))
44. labelsAndPoint = testData.map(lambda lp:lp.label).zip(predictions)
45. testMSE = labelsAndPoint.map(lambda (v,p):(v-p)**2).sum()/float(testData.count())
46. mylog.append("測試誤差是")
47. mylog.append(testMSE)
49. #存儲模型
51. model.save(sc,"/home/xiatao/machine_learing/")
52. sc.parallelize(mylog).saveAsTextFile("/home/xiatao/machine_learing/log")
53. sameModel = DecisionTreeModel.load(sc,"/home/xiatao/machine_learing/")