在 PySpark Dataframe 中使用自定义指标函数答案

【问题标题】：Use a custom metric function in PySpark Datafame在 PySpark Dataframe 中使用自定义指标函数
【发布时间】：2020-07-15 22:55:05
【问题描述】：

我已经在 python 中定义了一个自定义函数，以一种一对一的方式计算分类的 auc 分数。它以真实类别和不同类别的概率作为输入，并返回类别的 auc 分数。

from sklearn.metrics import roc_curve, auc
import pandas as pd

def mclass_auc(y_true, y_pred, n_class):
    tp = {}
    fp = {}
    aucs = {}
    for i in range(n_class):
    classes = [0]*n_class
    classes[i] = 1
    fp[i] tp[i], th = roc_curve(y_true.replace(list(range(n_class)), classes), y_pred[:, i])
    aucs[i] = auc(fp[i], tp[i])
    return aucs

为简单起见，我生成了一些总和不等于 1 的概率值。

cola = [np.random.randint(40, 81)/100 for i in range(10000)]
colb = [np.random.randint(30, 801)/1000 for i in range(10000)]
colc = [np.random.randint(40, 81)/200 for i in range(10000)]

coly = [np.random.randint(0, 4) for i in range(10000)]

sample_df = pd.DataFrame({'0':cola, '1':colb, '2':colc, 'y':coly})

y_true = sample_df['y']
y_pred = sample_df[['1','2','3']].values

auc_multiclass(y_true, y_pred, 3)
sql.createDataFrame(sample_df)

在python中，我可以使用上面的函数。有人可以帮我在 PySpark 数据框设置中计算吗？在这种情况下，将其更改为 pandas 数据框并计算失败。

【问题讨论】：

你试过udf了吗？您可以发布您尝试过的内容吗？
网上的例子都是基于sc.parallelize()。但就我而言，它不起作用。
你能发布你尝试过的东西吗？
stackoverflow.com/a/37587466/6597727 这是我试图复制的示例，但在 sc.parallelize(spark_df) 失败了

标签： python pandas pyspark scikit-learn

【解决方案1】：

from pyspark.mllib.evaluation import BinaryClassificationMetrics
import pyspark.sql.functions as F

def mclass_auc_spark(y_true, y_pred, n_class):
    aucs = {}
    for i in range(n_class):
        pred = y_pred.select(str(i)).withColumn('row_id', F.monotonically_increasing_id())
        true = y_true.withColumn('y', F.when(F.col('y') == i, 1.0).otherwise(0.0)).withColumn('row_id', F.monotonically_increasing_id())
        pred_labels = pred.join(true,on='row_id')
        metric = BinaryClassificationMetrics(pred_labels.select('y',str(i)).rdd)
        aucs[i] = metric.areaUnderROC
    return aucs

spark_df = sql.createDataFrame(sample_df)

y_true = spark_df.select('y')
y_pred = spark_df.select('0','1','2')

auc_scores = mclass_auc(y_true, y_pred, 3)

【讨论】：