如何使用线性回归评估 Spark Pipeline 中模型的性能（准确性）答案

【问题标题】：How to evaluate the performance of the model (accuracy) in Spark Pipeline with Linear Regression如何使用线性回归评估 Spark Pipeline 中模型的性能（准确性）
【发布时间】：2020-06-24 18:27:48
【问题描述】：

尝试使用线性回归运行 Spark 管道，我能够执行模型，并寻找

为了找到我需要模型摘要的模型效率和其他指标，我找到了一些 Python 示例，我已在下面评论以供参考。

       import org.apache.spark.ml.feature.VectorAssembler
       import spark.implicits._
       import org.apache.spark.sql
       import org.apache.spark.sql.functions._
       import org.apache.spark.sql.types.DecimalType
       import org.apache.spark.sql.{Dataset, Row, SparkSession}
       import org.apache.spark.ml.regression.LinearRegression
       import org.apache.spark.ml.feature.OneHotEncoderEstimator
       import org.apache.spark.ml.{Pipeline, PipelineModel}    

       val splitDF: Array[Dataset[Row]] = inputDF.randomSplit(Array(0.5, 0.5))
        val trainingDF = splitDF(0)
        val testingDF = splitDF(1) 


        val encoder = new OneHotEncoderEstimator()
          .setInputCols(Array("_LookUpID"))
          .setOutputCols(Array("_LookUpID_Encoded"))

        val requiredFeatures = Array("_LookUpID_Encoded","VALUE1")
        val assembler = new VectorAssembler()
          .setInputCols(requiredFeatures)
          .setOutputCol("features")


        val lr = new LinearRegression()
          .setMaxIter(10)
          .setRegParam(0.3)
          .setElasticNetParam(0.8)
          .setFeaturesCol("features")
          .setLabelCol("VALUE2")

        // Fit the model
        val pipeline = new Pipeline()
          .setStages(Array(encoder, assembler, lr))

        // Fit the pipeline to training documents.
        val lrModel = pipeline.fit(trainingDF)

        val predictions = lrModel.transform(testingDF)
        println("*** Predictions ***")
        predictions.printSchema()  

predictions.select("VALUE_DATE","_LookUpID","_CD","VALUE1","VALUE2","prediction").show(100)

        val rm = new RegressionMetrics(predictions.rdd.map(x => (x(4).asInstanceOf[Double], x(5).asInstanceOf[Double])))
        println("sqrt(MSE): " + Math.sqrt(rm.meanSquaredError))
        println("R Squared: " + rm.r2)
        println("Explained Variance: " + rm.explainedVariance + "\n")

使用分区摄取

def getDataFrame(sql: String, lowerNumber: Int, upperNumber: Int): DataFrame = {
 val inputDF: DataFrame = 
 spark.read.format(source = "jdbc")
  .option("url", "jdbc:oracle:thin:@//url")
        .option("user", "user")
        .option("password", "password")
        .option("driver", "oracle.jdbc.OracleDriver")
        .option("dbtable", s"($sql)")
        .option("partitionColumn", "_LookUpID")
        .option("numPartitions", "6")
        .option("lowerBound", lowerNumber)
        .option("upperBound", upperNumber)
        .load()
 inputDF
}

如果我提供一个包含 100 万行的数据集（在 100K 时可以正常工作），即使作业分配了 32GB 内存，以下 pipleline 内存不足（java.lang.OutOfMemoryError: Java heap space at...）。尝试 .cache() inputDF 没有太大成功。是因为编码_LookUpID，我还能做些什么不同更新：增加了驱动程序的堆内存以及分区数，并能够解决它。

谢谢

【问题讨论】：

您能否详细说明发生内存问题的位置？驱动还是在集群上？ spark-submit 选项是什么？ IE。 driver-memory 等。如果它抛出内存不足错误，可能是你真的没有足够的空间，你的 32 GB 设置不正确。或者可能只是您的数据集太大，您需要对其进行拆分。

标签： python scala apache-spark linear-regression apache-spark-ml

【解决方案1】：

用 RegressionMetrics 更新问题以获取 RMSE 和 R Squared 等指标

已分区数据集并增加了驱动程序的堆内存，目前已解决内存问题。会持续关注

【讨论】：