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

Question

我正在尝试提取我使用PySpark训练的随机森林对象的类概率.但是,我没有在文档中的任何地方看到它的示例,也不是它的方法RandomForestModel.

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

以下是文档中提供的示例代码,它仅提供最终类(而不是概率):

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()方法 - 我该怎么办？

Answer 1

据我所知,当前版本(1.2.1)不支持此功能.本机Scala代码(tree.py)上的Python包装器仅定义"预测"函数,而这些函数又调用相应的Scala对应函数(treeEnsembleModels.scala).后者通过在二元决策中进行投票来做出决策.更清晰的解决方案是提供概率预测,该预测可以任意阈值化或用于ROC计算,如sklearn.应该为将来的版本添加此功能!

作为一种解决方法,我将predict_proba实现为纯Python函数(参见下面的示例).它既不优雅也不高效,因为它在森林中的一组个体决策树上运行循环.诀窍 - 或者说是一个肮脏的黑客 - 是访问Java决策树模型的数组并将它们转换为Python对应物.之后,您可以计算整个数据集中的单个模型的预测,并使用"zip"在RDD中累积它们的总和.除以树的数量得到期望的结果.对于大型数据集,主节点中的少量决策树上的循环应该是可接受的.

由于将Python集成到Spark(在Java中运行)的困难,下面的代码相当棘手.应该非常小心,不要将任何复杂的数据发送到工作节点,这会因序列化问题导致崩溃.没有引用Spark上下文的代码可以在工作节点上运行.此外,不能序列化任何引用任何Java代码的代码.例如,在下面的代码中使用len(树)而不是ntree可能很诱人 - 砰!在Java/Scala中编写这样的包装器可以更加优雅,例如通过在工作节​​点上的决策树上运行循环,从而降低通信成本.

下面的测试函数表明predict_proba给出了与原始示例中使用的预测相同的测试错误.

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的好习惯!

Answer 2

这是现在可用的。

Spark ML 提供：

• apredictionCol包含预测标签
• 和一个probabilityCol包含每个标签概率的向量，这就是你要找的！
• 您还可以访问原始计数

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