PySpark和MLLib：随机森林预测的类概率

Question

PySpark和MLLib：随机森林预测的类概率

apache-sparkpysparkrandom-forestapache-spark-mllib

9

我正在尝试使用PySpark提取随机森林对象的类概率。然而，在文档中没有任何示例，也不是RandomForestModel的方法。

如何从PySpark的RandomForestModel分类器中提取类概率？

以下是文档中提供的示例代码，仅提供最终类别（而非概率）：

from pyspark.mllib.tree import RandomForest
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 RandomForest model.
#  Empty categoricalFeaturesInfo indicates all features are continuous.
#  Note: Use larger numTrees in practice.
#  Setting featureSubsetStrategy="auto" lets the algorithm choose.
model = RandomForest.trainClassifier(trainingData, numClasses=2, categoricalFeaturesInfo={},
                                     numTrees=3, featureSubsetStrategy="auto",
                                     impurity='gini', maxDepth=4, maxBins=32)

# Evaluate model on test instances and compute test error
predictions = model.predict(testData.map(lambda x: x.features))

我看不到model.predict_proba()方法 — 我该怎么办？

- Bryan

虽然有点晚，但是有一个使用Scala的分支解决方案： https://github.com/apache/spark/compare/master...mqk:master - Reuben L.

该问题现在已经在新的Spark ML库中得到（大部分）解决：https://dev59.com/i1cP5IYBdhLWcg3w6-IH#43643426 - desertnaut

4个回答

7

现在已经可用。

Spark ML 提供：

predictionCol ：包含预测标签的列
probabilityCol ：包含每个标签的概率向量，这就是你要找的东西！
您还可以访问原始计数。

更多详细信息，请参阅 Spark 文档： http://spark.apache.org/docs/latest/ml-classification-regression.html#output-columns-predictions

- Julien

确实 - 请参考以下示例：https://dev59.com/i1cP5IYBdhLWcg3w6-IH#43643426 - desertnaut

1

然而，它将与Spark 1.5.0和新的Spark-ML API一起提供。

- Olivier Girardot

0

也许大家已经看过这篇文章了，但今天我在尝试使用mllib计算多类分类器对训练集的准确率时遇到了同样的问题。因此，我想分享一下我的经验，希望能帮助到其他人...

可以通过以下简单的方法计算概率：

# say you have a testset against which you want to run your classifier
   (trainingset, testset) =data.randomSplit([0.7, 0.3])
   # I converted the spark dataset containing the test data to pandas
     ptd=testData.toPandas()

   #Now get a count of number of labels matching the predictions

   correct = ((ptd.label-1) == (predictions)).sum() 
   # here we had to change the labels from 0-9 as opposed to 1-10 since
   #labels take the values from 0 .. numClasses-1

   m=ptd.shape[0]
   print((correct/m)*100)

- sunny

网页内容由stack overflow 提供, 点击上面的

可以查看英文原文，
原文链接

- Pavel Laskov · Accepted Answer

据我所知，当前版本（1.2.1）不支持此功能。Python包装的Scala代码（tree.py）仅定义了“predict”函数，而这些函数又调用了各自对应的Scala版本（treeEnsembleModels.scala）。后者通过在二元决策之间进行投票来做出决策。更好的解决方案是提供可概率预测的功能，可任意进行阈值处理或用于ROC计算，就像sklearn中一样。这个特性应该在未来的版本中添加！作为一种变通方法，我实现了predict_proba作为一个纯Python函数（请参见下面的示例）。它既不优雅也不高效，因为它在森林中循环遍历了一组单独的决策树。技巧或者说是一种肮脏的hack，是访问Java决策树模型数组并将它们转换为Python模型。之后，您可以计算整个数据集上每个模型的预测，并使用“zip”在RDD中累加它们的总和。除以树的数量就得到了期望的结果。对于大型数据集，在主节点上循环遍历少量决策树应该是可以接受的。由于将Python集成到Spark（在Java中运行）中的困难，下面的代码相当棘手。你必须非常小心，不要将任何复杂的数据发送到worker节点，这会导致序列化问题而崩溃。不能在worker节点上运行任何涉及Spark上下文的代码。同样，不能序列化任何涉及Java代码的代码。例如，在下面的代码中使用len(trees)代替ntrees可能是很诱人的——砰！在Java/Scala中编写这样的包装器可能更加优雅，例如在worker节点上循环遍历决策树，从而减少通信成本。下面的测试函数证明了predict_proba给出的测试误差与原始示例中使用的predict相同。

def predict_proba(rf_model, data):
   '''
   This wrapper overcomes the "binary" nature of predictions in the native
   RandomForestModel. 
   '''

    # Collect the individual decision tree models by calling the underlying
    # Java model. These are returned as JavaArray defined by py4j.
    trees = rf_model._java_model.trees()
    ntrees = rf_model.numTrees()
    scores = DecisionTreeModel(trees[0]).predict(data.map(lambda x: x.features))

    # For each decision tree, apply its prediction to the entire dataset and
    # accumulate the results using 'zip'.
    for i in range(1,ntrees):
        dtm = DecisionTreeModel(trees[i])
        scores = scores.zip(dtm.predict(data.map(lambda x: x.features)))
        scores = scores.map(lambda x: x[0] + x[1])

    # Divide the accumulated scores over the number of trees
    return scores.map(lambda x: x/ntrees)

def testError(lap):
    testErr = lap.filter(lambda (v, p): v != p).count() / float(testData.count())
    print('Test Error = ' + str(testErr))


def testClassification(trainingData, testData):

    model = RandomForest.trainClassifier(trainingData, numClasses=2,
                                         categoricalFeaturesInfo={},
                                         numTrees=50, maxDepth=30)

    # Compute test error by thresholding probabilistic predictions
    threshold = 0.5
    scores = predict_proba(model,testData)
    pred = scores.map(lambda x: 0 if x < threshold else 1)
    lab_pred = testData.map(lambda lp: lp.label).zip(pred)
    testError(lab_pred)

    # Compute test error by comparing binary predictions
    predictions = model.predict(testData.map(lambda x: x.features))
    labelsAndPredictions = testData.map(lambda lp: lp.label).zip(predictions)
    testError(labelsAndPredictions)

总而言之，这是一个很好的练习，有助于学习Spark！