pyspark-ml学习笔记：模型评估

'''
模型评估模块：
· pyspark api
· sklearn api
'''

import numpy as np
from pyspark.ml.linalg import Vectors
from start_pyspark import spark, sc, sqlContext 
from pyspark.ml.evaluation import BinaryClassificationEvaluator

scoreAndLabels = map(lambda x: (Vectors.dense([1.0 - x[0], x[0]]), x[1]), [(0.1, 0.0), (0.1, 1.0), (0.4, 0.0), (0.6, 0.0), (0.6, 1.0), (0.6, 1.0), (0.8, 1.0)])
dataset = spark.createDataFrame(scoreAndLabels, ["prediction", "label"])
evaluator = BinaryClassificationEvaluator(rawPredictionCol="prediction")
res = evaluator.evaluate(dataset)
print ( 'res>>>>>>', res)

res = evaluator.evaluate(dataset, {evaluator.metricName: "areaUnderPR"})
print ( 'areaUnderPR>>>>>>', res)

res = evaluator.evaluate(dataset, {evaluator.metricName: "areaUnderROC"})
print ( 'areaUnderROC>>>>>>', res)


bb = dataset.select(dataset["prediction"])
print ('bbbbbb>>>>>', bb.collect() )

print ('rdd>>>>>', dataset.rdd.collect() )



pandas_pd = dataset.toPandas()
print ('bb>>>>>', pandas_pd )
import numpy as np
print ('bb>>>>>', pandas_pd['prediction'].values.tolist() )

print ('==========================================================================================================')
from sklearn.metrics import precision_score
y_true = np.array( [0.0, 1, 2, 0, 1, 2] )
print ('y_true:', y_true)
y_pred = np.array(  [0, 2, 1, 0, 0, 1] )
res = precision_score(y_true, y_pred, average='macro')  
print('precision_score:', res)


print ('==========================================================================================================')
from sklearn.metrics import precision_score, recall_score, roc_auc_score, roc_curve, \
    accuracy_score, r2_score, f1_score

y_pred = np.array( [ np.array ( per_pd ) for per_pd in pandas_pd['prediction'].values ] )
print ( 'y_pred:', y_pred )
y = np.array( [ np.array ( per_pd ) for per_pd in pandas_pd['label'].values ] )
print ( 'y:', y )
print ( 'y_pred[:, 1]:',  y_pred[:, 1] )
# 需要概率值。
roc_auc_score_res = roc_auc_score(y, y_pred[:, 1])
print('roc_auc_score_res:', roc_auc_score_res)


def prob2label(prob, thres = 0.5):
    prob[ prob>thres ] = 1.0
    prob[ prob==thres ] = 1.0 
    prob[ prob<thres ] = 0.0 
    return prob

# 需要label值。
precision_score_res = precision_score(y, prob2label( y_pred[:, 1] ) )
print('precision_score_res:', precision_score_res)

recall_score_res = recall_score(y, prob2label( y_pred[:, 1] ) )
print('recall_score_res:', recall_score_res)

roc_curve_res = roc_curve(y, prob2label( y_pred[:, 1] ) )
print('roc_curve_res:', roc_curve_res)

accuracy_score_res = accuracy_score(y, prob2label( y_pred[:, 1] ) )
print('accuracy_score_res:', accuracy_score_res)

r2_score_res = r2_score(y, prob2label( y_pred[:, 1] ) )
print('r2_score_res:', r2_score_res)

f1_score_res = f1_score(y, prob2label( y_pred[:, 1] ) )
print('f1_score_res:', f1_score_res)

import os
import sys
 
''' 
#下面这些目录都是你自己机器的Spark安装目录和Java安装目录
os.environ['SPARK_HOME'] = "/Users/***/spark-2.4.3-bin-hadoop2.7/" 
sys.path.append("/Users/***/spark-2.4.3-bin-hadoop2.7/bin") 
sys.path.append("/Users/***/spark-2.4.3-bin-hadoop2.7/python") 
sys.path.append("/Users/***/spark-2.4.3-bin-hadoop2.7/python/pyspark") 
sys.path.append("/Users/***/spark-2.4.3-bin-hadoop2.7/python/lib") 
sys.path.append("/Users/***/spark-2.4.3-bin-hadoop2.7/python/lib/pyspark.zip") 
sys.path.append("/Users/***/spark-2.4.3-bin-hadoop2.7/lib/py4j-0.9-src.zip") 
# sys.path.append("/Library/Java/JavaVirtualMachines/jdk1.8.0_144.jdk/Contents/Home") 
os.environ['JAVA_HOME'] = "/Library/Java/JavaVirtualMachines/jdk1.8.0_181.jdk/Contents/Home" 
'''
 
from pyspark.sql import SparkSession, SQLContext
from pyspark import SparkConf, SparkContext 
#conf = SparkConf().setMaster("local").setAppName("My App") 
conf = SparkConf().setMaster("yarn").setAppName("My App") 
sc = SparkContext(conf = conf) 
spark = SparkSession.builder.appName('CalculatingGeoDistances').getOrCreate() 
sqlContext = SQLContext(sparkContext=sc)