model.predictProbabilities() для логистической регрессии в Spark?

Есть два способа выполнить логистическую регрессию в Spark: spark.ml и spark.mllib.

С DataFrames вы можете использовать spark.ml:

import org.apache.spark
import sqlContext.implicits._

def p(label: Double, a: Double, b: Double) =
  new spark.mllib.regression.LabeledPoint(
    label, new spark.mllib.linalg.DenseVector(Array(a, b)))

val data = sc.parallelize(Seq(p(1.0, 0.0, 0.5), p(0.0, 0.5, 1.0)))
val df = data.toDF

val model = new spark.ml.classification.LogisticRegression().fit(df)
model.transform(df).show

Вы получаете необработанные прогнозы и вероятности:

+-----+---------+--------------------+--------------------+----------+
|label| features|       rawPrediction|         probability|prediction|
+-----+---------+--------------------+--------------------+----------+
|  1.0|[0.0,0.5]|[-19.037302860930...|[5.39764620520461...|       1.0|
|  0.0|[0.5,1.0]|[18.9861466274786...|[0.99999999431904...|       0.0|
+-----+---------+--------------------+--------------------+----------+

С RDD вы можете использовать spark.mllib:

val model = new spark.mllib.classification.LogisticRegressionWithLBFGS().run(data)

Эта модель не предоставляет необработанных прогнозов и вероятностей. Вы можете взглянуть на predictPoint. Он умножает векторы и выбирает класс с самым высоким прогнозом. Веса общедоступны, поэтому вы можете скопировать этот алгоритм и сохранить прогнозы, а не просто возвращать самый высокий.

Следуя предложениям @Daniel Darabos:

1. Я попытался использовать функцию LogisticRegression из ml вместо mllib. К сожалению, она не поддерживает мультиклассовую логистическую регрессию, а только бинарную.
2. Я взглянул на PredictedPoint и изменил его так, чтобы он печатал все вероятности для каждого класса. Вот как это выглядит:

    def predictPointForMulticlass(featurizedVector:Vector,weightsArray:Vector,intercept:Double,numClasses:Int,numFeatures:Int) : Seq[(String, Double)] = {

        val weightsArraySize = weightsArray.size
        val dataWithBiasSize = weightsArraySize / (numClasses - 1)
        val withBias = false

        var bestClass = 0
        var maxMargin = 0.0
        var margins = new Array[Double](numClasses - 1)
        var temp_marginMap = new HashMap[Int, Double]()
        var res = new HashMap[Int, Double]()

        (0 until numClasses - 1).foreach { i =>
          var margin = 0.0
          var index = 0
          featurizedVector.toArray.foreach(value => {
            if (value != 0.0) {
              margin += value * weightsArray((i * dataWithBiasSize) + index)
            }
            index += 1
          }
          )
        // Intercept is required to be added into margin.
        if (withBias) {
            margin += weightsArray((i * dataWithBiasSize) + featurizedVector.size)
        }
        val prob = 1.0 / (1.0 + Math.exp(-margin))
        margins(i) = margin

        temp_marginMap += (i -> margin)

        if(margin > maxMargin) {
            maxMargin = margin
            bestClass = i + 1
          }
        }

        for ((k,v) <- temp_marginMap){
          val calc =probCalc(maxMargin,v)
          res += (k -> calc)
        }

      return res
    }

где probCalc() просто определяется как:

      def probCalc(maxMargin:Double,margin:Double) :Double ={
        val res = 1.0 / (1.0 + Math.exp(-(margin - maxMargin)))
        res
      }

Я возвращаю Hashmap[Int, Double], но это можно изменить по вашему желанию.

Надеюсь, это поможет!