from pyspark import SparkConf
from pyspark.sql import SparkSession
from pyspark.ml.linalg import Vectors
from pyspark.ml.feature import StringIndexer
from pyspark.ml.classification import RandomForestClassifier
from pyspark.sql import Row
import pandas as pd
from sklearn import metrics
 
if __name__ == "__main__":
  appname = "RandomForestClassifier"
  master ="local[4]" 
  conf = SparkConf().setAppName(appname).setMaster(master) #spark配置        
  spark=SparkSession.builder.config(conf=conf).getOrCreate()#spark實例化
  
#讀取數據
  data=spark.read.csv('良惡性乳腺癌數據.csv',header=True)
  
#構造訓練數據集
  dataSet = data.na.fill('0').rdd.map(list)#用0填充空值  
  trainData, testData= dataSet.randomSplit([0.7, 0.3], seed=7)
  trainingSet = trainData.map(lambda x:Row(label=x[-1], features=Vectors.dense(x[:-1]))).toDF()  
  train_num = trainingSet.count()
  print("訓練樣本數:{}".format(train_num))
 
   
#使用隨機森林進行訓練
  stringIndexer = StringIndexer(inputCol="label", outputCol="indexed")
  si_model = stringIndexer.fit(trainingSet)
  train_tf = si_model.transform(trainingSet)
  train_tf.show(5)  
  rf = RandomForestClassifier(numTrees=100, labelCol="indexed", seed=7)
  rfModel = rf.fit(train_tf)
   
#輸出模型特征重要性、子樹權重
  print("模型特征重要性:{}".format(rfModel.featureImportances))
  print("模型特征數:{}".format(rfModel.numFeatures))
  
#預測測試集
  testSet = testData.map(lambda x:Row(label=x[-1], features=Vectors.dense(x[:-1]))).toDF()
  test_num=testSet.count()
  print("測試樣本數:{}".format(test_num))  
  si_model = stringIndexer.fit(testSet)
  test_tf = si_model.transform(testSet)  
  predictResult = rfModel.transform(test_tf)
  predictResult.show(5)
  spark.stop()
 
#將預測結果轉為python中的dataframe
  columns=predictResult.columns#提取強表字段
  predictResult=predictResult.take(test_num)#
  predictResult=pd.DataFrame(predictResult,columns=columns)#轉為python中的dataframe
 
#性能評估
  y=list(predictResult['indexed'])
  y_pred=list(predictResult['prediction'])
  y_predprob=[x[1] for x in list(predictResult['probability'])]
  precision_score=metrics.precision_score(y, y_pred)#精確率
  recall_score=metrics.recall_score(y, y_pred)#召回率
  accuracy_score=metrics.accuracy_score(y, y_pred)#準確率
  f1_score=metrics.f1_score(y, y_pred)#F1分數
  auc_score=metrics.roc_auc_score(y, y_predprob)#auc分數
  print("精確率:",precision_score )#精確率
  print("召回率:",recall_score )#召回率
  print("準確率:",accuracy_score )#準確率
  print("F1分數:", f1_score)#F1分數
  print("auc分數:",auc_score )#auc分數