pyspark提取ROC曲线？

Question

有没有办法从 pyspark 中的 Spark ML 获取 ROC 曲线上的点？在文档中，我看到了 Scala 但不是 python 的示例：https://spark.apache.org/docs/2.1.0/mllib-evaluation-metrics.html

对吗？我当然可以想到实现它的方法，但我必须想象如果有一个预先构建的功能会更快。我正在处理 300 万个分数和几十个模型，所以速度很重要。

Answer 1

只要 ROC 曲线是 FPR 与 TPR 的图，就可以提取所需的值如下：

your_model.summary.roc.select('FPR').collect()
your_model.summary.roc.select('TPR').collect())

your_model 可以是例如您从以下内容中获得的模型：

from pyspark.ml.classification import LogisticRegression
log_reg = LogisticRegression()
your_model = log_reg.fit(df)

现在您应该只针对 TPR 绘制 FPR，例如使用 matplotlib。

附言

这是一个使用名为your_model（以及其他任何东西！）的模型绘制 ROC 曲线的完整示例。我还在 ROC 图中绘制了一条参考“随机猜测”线。

import matplotlib.pyplot as plt
plt.figure(figsize=(5,5))
plt.plot([0, 1], [0, 1], 'r--')
plt.plot(your_model.summary.roc.select('FPR').collect(),
         your_model.summary.roc.select('TPR').collect())
plt.xlabel('FPR')
plt.ylabel('TPR')
plt.show()

Answer 2

对于除逻辑回归之外适用于模型的更通用解决方案（例如缺少模型摘要的决策树或随机森林），您可以使用 Spark MLlib 中的BinaryClassificationMetrics 获取 ROC 曲线。

请注意，PySpark 版本并未实现 Scala version 所做的所有方法，因此您需要使用来自 JavaModelWrapper 的 .call(name) 函数。看来py4j也不支持解析scala.Tuple2类，所以必须手动处理。

例子：

from pyspark.mllib.evaluation import BinaryClassificationMetrics

# Scala version implements .roc() and .pr()
# Python: https://spark.apache.org/docs/latest/api/python/_modules/pyspark/mllib/common.html
# Scala: https://spark.apache.org/docs/latest/api/java/org/apache/spark/mllib/evaluation/BinaryClassificationMetrics.html
class CurveMetrics(BinaryClassificationMetrics):
    def __init__(self, *args):
        super(CurveMetrics, self).__init__(*args)

    def _to_list(self, rdd):
        points = []
        # Note this collect could be inefficient for large datasets 
        # considering there may be one probability per datapoint (at most)
        # The Scala version takes a numBins parameter, 
        # but it doesn't seem possible to pass this from Python to Java
        for row in rdd.collect():
            # Results are returned as type scala.Tuple2, 
            # which doesn't appear to have a py4j mapping
            points += [(float(row._1()), float(row._2()))]
        return points

    def get_curve(self, method):
        rdd = getattr(self._java_model, method)().toJavaRDD()
        return self._to_list(rdd)

用法：

import matplotlib.pyplot as plt

# Create a Pipeline estimator and fit on train DF, predict on test DF
model = estimator.fit(train)
predictions = model.transform(test)

# Returns as a list (false positive rate, true positive rate)
preds = predictions.select('label','probability').rdd.map(lambda row: (float(row['probability'][1]), float(row['label'])))
points = CurveMetrics(preds).get_curve('roc')

plt.figure()
x_val = [x[0] for x in points]
y_val = [x[1] for x in points]
plt.title(title)
plt.xlabel(xlabel)
plt.ylabel(ylabel)
plt.plot(x_val, y_val)

Scala 中的 BinaryClassificationMetrics 还实现了其他几个有用的方法：

metrics = CurveMetrics(preds)
metrics.get_curve('fMeasureByThreshold')
metrics.get_curve('precisionByThreshold')
metrics.get_curve('recallByThreshold')

Answer 3

要获得训练数据（训练模型）的 ROC 指标，我们可以使用 your_model.summary.roc，这是一个包含 FPR 和 TPR 列的 DataFrame。请参阅 Andrea 的回答。

对于在任意测试数据上评估的 ROC，我们可以使用 label 和 probability 列传递给 sklearn 的 roc_curve 以获取 FPR 和 TPR。这里我们假设一个二元分类问题，其中 y 分数是预测为 1 的概率。另见How to split Vector into columns - using PySpark、How to convert a pyspark dataframe column to numpy array

例子

from sklearn.metrics import roc_curve

model = lr.fit(train_df)
test_df_predict = model.transform(test_df)

y_score = test_df_predict.select(vector_to_array("probability")[1]).rdd.keys().collect()
y_true = test_df_predict.select("label").rdd.keys().collect()
fpr, tpr, thresholds = roc_curve(y_true, y_score)