apache-spark-mllib - Spark LinearRegressionSummary“正常”摘要

根据LinearRegressionSummary (Spark 2.1.0 JavaDoc)，p值仅可用于“正常”求解器。


该值仅在使用“普通”求解器时可用。


什么是“正常”求解器？

我正在这样做：

import org.apache.spark.ml.{Pipeline, PipelineModel}
import org.apache.spark.ml.evaluation.RegressionEvaluator
import org.apache.spark.ml.feature.VectorAssembler
import org.apache.spark.ml.regression.LinearRegressionModel
import org.apache.spark.ml.tuning.{CrossValidator, CrossValidatorModel, ParamGridBuilder}
import org.apache.spark.sql.functions._
import org.apache.spark.sql.{DataFrame, SparkSession}
    .
    .
    .
val (trainingData, testData): (DataFrame, DataFrame) =
  com.acme.pta.accuracy.Util.splitData(output, testProportion)
    .
    .
    .
val lr =
  new org.apache.spark.ml.regression.LinearRegression()
  .setSolver("normal").setMaxIter(maxIter)

val pipeline = new Pipeline()
  .setStages(Array(lr))

val paramGrid = new ParamGridBuilder()
  .addGrid(lr.elasticNetParam, Array(0.2, 0.4, 0.8, 0.9))
  .addGrid(lr.regParam, Array(0,6, 0.3, 0.1, 0.01))
  .build()

val cv = new CrossValidator()
  .setEstimator(pipeline)
  .setEvaluator(evaluator)
  .setEstimatorParamMaps(paramGrid)
  .setNumFolds(numFolds) // Use 3+ in practice

val cvModel: CrossValidatorModel = cv.fit(trainingData)

val pipelineModel: PipelineModel = cvModel.bestModel.asInstanceOf[PipelineModel]
val lrModel: LinearRegressionModel =
  pipelineModel.stages(0).asInstanceOf[LinearRegressionModel]

val modelSummary = lrModel.summary
Holder.log.info("lrModel.summary: " + modelSummary)
try {
  Holder.log.info("feature p values: ")
  // Exception occurs on line below.
  val featuresAndPValues = features.zip(lrModel.summary.pValues)
  featuresAndPValues.foreach(
    (featureAndPValue: (String, Double)) =>
    Holder.log.info(
      "feature: " + featureAndPValue._1 + ": " + featureAndPValue._2))
} catch {
  case _: java.lang.UnsupportedOperationException
            => Holder.log.error("Cannot compute p-values")
}

我仍在获取UnsupportedOperationException。

异常消息是：


没有适用于此LinearRegressionModel的p值


还有什么我需要做的吗？我正在使用

  "org.apache.spark" %% "spark-mllib" % "2.1.1"

该版本支持pValues吗？

最佳答案

更新

tl; dr

解决方案1

在普通LinearRegression中，仅当参数elasticNetParam或regParam之一为零时，才会显示pValue和其他“常规”统计信息。所以你可以改变

.addGrid( lr.elasticNetParam, Array( 0.0 ) )

要么

.addGrid( lr.regParam, Array( 0.0 ) )

解决方案2

制作LinearRegression的自定义版本，该版本将明确使用


用于回归的“常规”求解器。
Cholesky的WeightedLeastSquares求解器。


我将此类作为对ml.regression包的扩展。

package org.apache.spark.ml.regression

import scala.collection.mutable

import org.apache.spark.SparkException
import org.apache.spark.internal.Logging
import org.apache.spark.ml.feature.Instance
import org.apache.spark.ml.linalg.{Vector, Vectors}
import org.apache.spark.ml.optim.WeightedLeastSquares
import org.apache.spark.ml.param.{Param, ParamMap, ParamValidators}
import org.apache.spark.ml.util._
import org.apache.spark.mllib.linalg.VectorImplicits._
import org.apache.spark.rdd.RDD
import org.apache.spark.sql.{DataFrame, Dataset, Row}
import org.apache.spark.sql.functions._

class CholeskyLinearRegression ( override val uid: String )
    extends Regressor[ Vector, CholeskyLinearRegression, LinearRegressionModel ]
    with LinearRegressionParams with DefaultParamsWritable with Logging {

    import CholeskyLinearRegression._

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

    def setRegParam(value: Double): this.type = set(regParam, value)
    setDefault(regParam -> 0.0)

    def setFitIntercept(value: Boolean): this.type = set(fitIntercept, value)
    setDefault(fitIntercept -> true)

    def setStandardization(value: Boolean): this.type = set(standardization, value)
    setDefault(standardization -> true)

    def setElasticNetParam(value: Double): this.type = set(elasticNetParam, value)
    setDefault(elasticNetParam -> 0.0)

    def setMaxIter(value: Int): this.type = set(maxIter, value)
    setDefault(maxIter -> 100)

    def setTol(value: Double): this.type = set(tol, value)
    setDefault(tol -> 1E-6)

    def setWeightCol(value: String): this.type = set(weightCol, value)

    def setSolver(value: String): this.type = set(solver, value)
    setDefault(solver -> Auto)

    def setAggregationDepth(value: Int): this.type = set(aggregationDepth, value)
    setDefault(aggregationDepth -> 2)

    override protected def train(dataset: Dataset[_]): LinearRegressionModel = {

        // Extract the number of features before deciding optimization solver.
        val numFeatures = dataset.select(col($(featuresCol))).first().getAs[Vector](0).size
        val w = if (!isDefined(weightCol) || $(weightCol).isEmpty) lit(1.0) else col($(weightCol))

        val instances: RDD[Instance] =
            dataset
            .select( col( $(labelCol) ), w, col( $(featuresCol) ) )
            .rdd.map {
                case Row(label: Double, weight: Double, features: Vector) =>
                Instance(label, weight, features)
            }

        // if (($(solver) == Auto &&
        //   numFeatures <= WeightedLeastSquares.MAX_NUM_FEATURES) || $(solver) == Normal) {
        // For low dimensional data, WeightedLeastSquares is more efficient since the
        // training algorithm only requires one pass through the data. (SPARK-10668)

        val optimizer = new WeightedLeastSquares(
            $(fitIntercept),
            $(regParam),
            elasticNetParam = $(elasticNetParam),
            $(standardization),
            true,
            solverType = WeightedLeastSquares.Cholesky,
            maxIter = $(maxIter),
            tol = $(tol)
        )

        val model = optimizer.fit(instances)

        val lrModel = copyValues(new LinearRegressionModel(uid, model.coefficients, model.intercept))
        val (summaryModel, predictionColName) = lrModel.findSummaryModelAndPredictionCol()

        val trainingSummary = new LinearRegressionTrainingSummary(
            summaryModel.transform(dataset),
            predictionColName,
            $(labelCol),
            $(featuresCol),
            summaryModel,
            model.diagInvAtWA.toArray,
            model.objectiveHistory
        )

        lrModel
        .setSummary( Some( trainingSummary ) )

        lrModel
    }

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

object CholeskyLinearRegression
    extends DefaultParamsReadable[CholeskyLinearRegression] {

    override def load(path: String): CholeskyLinearRegression = super.load(path)

    val MAX_FEATURES_FOR_NORMAL_SOLVER: Int = WeightedLeastSquares.MAX_NUM_FEATURES

    /** String name for "auto". */
    private[regression] val Auto = "auto"

    /** String name for "normal". */
    private[regression] val Normal = "normal"

    /** String name for "l-bfgs". */
    private[regression] val LBFGS = "l-bfgs"

    /** Set of solvers that LinearRegression supports. */
    private[regression] val supportedSolvers = Array(Auto, Normal, LBFGS)
}

您所要做的就是将其粘贴到项目中的单独文件中，然后在代码中将LinearRegression更改为CholeskyLinearRegression。

val lr = new CholeskyLinearRegression() // new LinearRegression()
        .setSolver( "normal" )
        .setMaxIter( maxIter )

它适用于非零参数，并提供pValues。在以下参数网格上进行了测试。

val paramGrid = new ParamGridBuilder()
        .addGrid( lr.elasticNetParam, Array( 0.2, 0.4, 0.8, 0.9 ) )
        .addGrid( lr.regParam, Array( 0.6, 0.3, 0.1, 0.01 ) )
        .build()

全面调查

我最初以为主要问题是模型没有完全保留。装入CrossValidator后，不会保留训练后的模型。由于内存消耗，这是可以理解的。关于如何解决该问题，目前正在进行中。 JIRA中的debate。

您可以在注释部分中看到，我试图从最佳模型中提取参数以便再次运行。然后我发现模型摘要是可以的，只是对于某些参数diagInvAtWa的长度为1，基本上为零。

对于岭回归或Tikhonov正则化（elasticNet = 0）以及任何regParam pValues和其他“正常”统计量都可以计算，但是对于Lasso方法，介于两者之间的值（弹性网）不能计算。 regParam = 0也是如此：使用任何elasticNet都会计算pValue。

这是为什么

LinearRegression Issue加权最小二乘优化器，用于带有solverType = WeightedLeastSquares.Auto的“常规”求解器。该优化器的求解器为uses：QuasiNewton或Cholesky。仅当regParam和elasticNetParam均为非零时，才选择前者。

val solver = if (
    ( solverType == WeightedLeastSquares.Auto &&
        elasticNetParam != 0.0 &&
        regParam != 0.0 ) ||
    ( solverType == WeightedLeastSquares.QuasiNewton ) ) {

    ...
    new QuasiNewtonSolver(fitIntercept, maxIter, tol, effectiveL1RegFun)
} else {
    new CholeskySolver
}

因此，在参数网格中将始终使用QuasiNewtonSolver，因为regParam和elasticNetParam没有组合，其中之一为零。

我们知道，为了获得pValues和其他“正常”统计信息，例如t统计或std。系数的误差矩阵对角线（A ^ T * W * A）^-1（diagInvAtWA）不得是只有一个零的向量。此条件在two options的定义中设置。

diagInvAtWA是压缩的上三角矩阵（solution.aaInv）的对角元素的向量。

val diagInvAtWA = solution.aaInv.map { inv => ...

对于Cholesky solver，它是pValues，但是对于QuasiNewton calculated。 NormalEquationSolution的第二个not是此矩阵。

从技术上讲，您可以使用以下方法制作自己的LinearRegression版本

再生产

在此示例中，我使用了来自parameter的数据sample_linear_regression_data.txt。

完整的复制代码

import org.apache.spark._

import org.apache.spark.ml.{Pipeline, PipelineModel}
import org.apache.spark.ml.evaluation.{RegressionEvaluator, BinaryClassificationEvaluator}
import org.apache.spark.ml.feature.VectorAssembler
import org.apache.spark.ml.regression.{LinearRegressionModel, LinearRegression}
import org.apache.spark.ml.tuning.{CrossValidator, CrossValidatorModel, ParamGridBuilder}
import org.apache.spark.sql.functions._
import org.apache.spark.sql.{DataFrame, SparkSession}
import org.apache.spark.ml.param.ParamMap

object Main {

    def main( args: Array[ String ] ): Unit = {

        val spark =
            SparkSession
            .builder()
            .appName( "SO" )
            .master( "local[*]" )
            .config( "spark.driver.host", "localhost" )
            .getOrCreate()

        import spark.implicits._

        val data =
            spark
            .read
            .format( "libsvm" )
            .load( "./sample_linear_regression_data.txt" )

        val Array( training, test ) =
            data
            .randomSplit( Array( 0.9, 0.1 ), seed = 12345 )

        val maxIter = 10;

        val lr = new LinearRegression()
            .setSolver( "normal" )
            .setMaxIter( maxIter )

        val paramGrid = new ParamGridBuilder()
            // .addGrid( lr.elasticNetParam, Array( 0.2, 0.4, 0.8, 0.9 ) )
            .addGrid( lr.elasticNetParam, Array( 0.0 ) )
            .addGrid( lr.regParam, Array( 0.6, 0.3, 0.1, 0.01 ) )
            .build()

        val pipeline = new Pipeline()
            .setStages( Array( lr ) )

        val cv = new CrossValidator()
            .setEstimator( pipeline )
            .setEvaluator( new RegressionEvaluator )
            .setEstimatorParamMaps( paramGrid )
            .setNumFolds( 2 )  // Use 3+ in practice

        val cvModel =
            cv
            .fit( training )

        val pipelineModel: PipelineModel =
            cvModel
            .bestModel
            .asInstanceOf[ PipelineModel ]

        val lrModel: LinearRegressionModel =
            pipelineModel
            .stages( 0 )
            .asInstanceOf[ LinearRegressionModel ]

        // Technically there is a way to use exact ParamMap
        // to build a new LR but for the simplicity I'll
        // get and set them explicitly

        // lrModel.params.foreach( ( param ) => {

        //     println( param )
        // } )

        // val bestLr = new LinearRegression()
        //     .setSolver( "normal" )
        //     .setMaxIter( maxIter )
        //     .setRegParam( lrModel.getRegParam )
        //     .setElasticNetParam( lrModel.getElasticNetParam )

        // val bestLrModel = bestLr.fit( training )

        val modelSummary =
            lrModel
            .summary

        println( "lrModel pValues: " + modelSummary.pValues.mkString( ", " ) )

        spark.stop()
    }
}

原版的

共有三种求解器算法here：


l-bfgs-受限内存的Broyden–Fletcher–Goldfarb–Shanno算法，它是有限内存的准牛顿优化available。
normal-使用method作为线性回归问题的解析解决方案。它基本上是Normal Equation或weighted least squares approach。
auto-自动选择求解器算法。可能时将使用法线方程求解器，但是在需要时，它将自动使用迭代优化方法


coefficientStandardErrors，tValues和pValues仅在使用“标准”求解器时可用，因为它们都基于diagInvAtWA-矩阵（A ^ T * W * A）^-1的对角线。

关于apache-spark-mllib - Spark LinearRegressionSummary“正常”摘要，我们在Stack Overflow上找到一个类似的问题：https://stackoverflow.com/questions/46696378/