Spark MLLIB: вычислить значение, подобное stddev, для регрессии случайных лесов

Question

У меня есть некоторые данные, по которым я хочу изучить «нормальное» поведение.

Используя ограниченный набор переменных, мне удалось сделать это с простым средним значением.

df.groupBy([My_Variables]) 
  .agg(
       mean("value").alias("prediction"),
       stddev("value").alias("sigma")
  )

Примечание: "value" - это двойное поле

Я также сделал то же самое, используя алгоритм Random Forest, который позволяет мне использовать больше переменных.

val limit_training_set:Long = 1517439600

val trainingData = df.filter(col("datetime").cast("long")<limit_training_set)
val testData = df.filter(col("datetime").cast("long")>limit_training_set)


val assembler = new VectorAssembler()
      .setInputCols(Array(
        [My_Variables]
      ))
      .setOutputCol("features")

... // (define Indexers and Imputers)

val rf = new RandomForestRegressor()
  .setNumTrees(10) 
  .setMaxDepth(18) 
  .setLabelCol("value")
  .setFeaturesCol("features")

val pipeline = new Pipeline()
    .setStages(Array([Indexers and Imputers], assembler, rf))


val paramGrid = new ParamGridBuilder()
  .addGrid(rf.numTrees, Array(5,10))
  .addGrid(rf.maxDepth, Array(10,18)) 
  .build()

// Set up cross-validation.
val re = new RegressionEvaluator()
  .setMetricName("mae")
  .setLabelCol("value")

val tv = new TrainValidationSplit()
  .setEstimator(pipeline)
  .setEvaluator(re)
  .setEstimatorParamMaps(paramGrid)
  // 80% of the data will be used for training and the remaining 20% for validation.
  .setTrainRatio(0.8)


val model = tv.fit(trainingData)

Это дает мне довольно хорошеепредсказания, но по сравнению с методом Mean я теряю информацию о стандартном отклонении, которую я хотел бы иметь.

Есть ли способ вычисления значения, подобного stddev, с использованием Random Forest в дополнение к предсказанию?Или есть другой алгоритм ML, который подходит для этого лучше?

Answer 1

Вы можете добавить UnaryTransformer, чтобы вычислить счет желаемых полей.Это добавит новое поле к вашей строке:

class ScoreField(override val uid: String)
  extends UnaryTransformer[Double, Double, ScoreField]
    with DefaultParamsWritable {

  def this() = this(Identifiable.randomUID("Std"))

  final val mean: DoubleParam = new DoubleParam(this, "mean", "mean")

  final val std: DoubleParam = new DoubleParam(this, "std", "std")

  def setMean(m: Double) = set(mean, m)

  def setStd(s: Double) = set(std, s)

  override protected def createTransformFunc: Double => Double =
    v => { (v - $ { mean } / $ { std }) }

  override protected def outputDataType: DataType = DoubleType

  override def copy(extra: ParamMap): ScoreField = defaultCopy(extra)
}

object ScoreField extends DefaultParamsReadable[ScoreField] {
  def apply() : ScoreField = new ScoreField()
  override def load(path: String): ScoreField = super.load(path)
}

// Create each stage for each field 
val score = new ScoreField()
score.setInputCol("inputField")
score.setOutputCol("outputField")
// Add your mean and std four your field, this must be executed previously
score.setMean(4.0)
score.setStd(2.0)

Вы должны добавить в свой конвейер:

val pipeline = new Pipeline()
.setStages(Array([Indexers and Imputers], score, assembler, rf))

Как только вы вызовете transform для вашего конвейера, вы получите строку сваш прогноз и ваши забитые поля.

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