ользую Spark, и я хотел бы обучить модели машинного обучения.

Из-за плохих результатов я хотел бы отобразить ошибку, допущенную моделью в каждую эпоху обучения (в обучении и наборе тестовых данных).

Затем я буду использовать эту информацию, чтобы определить, не соответствует ли моя модель данным.

Вопрос: Как я могу нарисовать кривую обучения модели с искрой?

В следующем примере я реализовал свой собственный оценщик и переопределил метод оценки, чтобы напечатать необходимые мне показатели, но были показаны только два значения (maxIter = 1000).

MinimalRunnableCode.scala:

import org.apache.spark.SparkConf
import org.apache.spark.ml.linalg.Vectors
import org.apache.spark.ml.regression.LinearRegression
import org.apache.spark.ml.tuning.{ParamGridBuilder, TrainValidationSplit}
import org.apache.spark.sql.SparkSession

object Min extends App {

  // Open spark session.
  val conf = new SparkConf()
    .setMaster("local")
    .set("spark.network.timeout", "800")

  val ss = SparkSession.builder
    .config(conf)
    .getOrCreate

  // Load data.
  val data = ss.createDataFrame(ss.sparkContext.parallelize(
      List(
        (Vectors.dense(1, 2), 1),
        (Vectors.dense(1, 3), 2),
        (Vectors.dense(1, 2), 1),
        (Vectors.dense(1, 3), 2),
        (Vectors.dense(1, 2), 1),
        (Vectors.dense(1, 3), 2),
        (Vectors.dense(1, 2), 1),
        (Vectors.dense(1, 3), 2),
        (Vectors.dense(1, 2), 1),
        (Vectors.dense(1, 3), 2),
        (Vectors.dense(1, 4), 3)
      )
    ))
    .withColumnRenamed("_1", "features")
    .withColumnRenamed("_2", "label")

  val Array(training, test) = data.randomSplit(Array(0.8, 0.2), seed = 42)

  // Create model of linear regression.
  val lr = new LinearRegression().setMaxIter(1000)

  // Create parameters grid that will be used to train different version of the linear model.
  val paramGrid = new ParamGridBuilder()
    .addGrid(lr.regParam, Array(0.001))
    .addGrid(lr.fitIntercept)
    .addGrid(lr.elasticNetParam, Array(0.5))
    .build()

  // Create trainer using validation split to evaluate which set of parameters performs the best.
  val trainValidationSplit = new TrainValidationSplit()
    .setEstimator(lr)
    .setEvaluator(new CustomRegressionEvaluator)
    .setEstimatorParamMaps(paramGrid)
    .setTrainRatio(0.8) // 80% of the data will be used for training and the remaining 20% for validation.

  // Run train validation split, and choose the best set of parameters.
  var model = trainValidationSplit.fit(training)

  // Close spark session.
  ss.stop()
}

CustomRegressionEvaluator.scala:

import org.apache.spark.ml.evaluation.{Evaluator, RegressionEvaluator}
import org.apache.spark.ml.param.{Param, ParamMap, Params}
import org.apache.spark.ml.util.{DefaultParamsReadable, DefaultParamsWritable, Identifiable}
import org.apache.spark.mllib.evaluation.RegressionMetrics
import org.apache.spark.sql.{Dataset, Row}
import org.apache.spark.sql.functions._
import org.apache.spark.sql.types._

final class CustomRegressionEvaluator (override val uid: String) extends Evaluator with HasPredictionCol with HasLabelCol with DefaultParamsWritable {

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

  def checkNumericType(
                        schema: StructType,
                        colName: String,
                        msg: String = ""): Unit = {
    val actualDataType = schema(colName).dataType
    val message = if (msg != null && msg.trim.length > 0) " " + msg else ""
    require(actualDataType.isInstanceOf[NumericType], s"Column $colName must be of type " +
      s"NumericType but was actually of type $actualDataType.$message")
  }

  def checkColumnTypes(
                        schema: StructType,
                        colName: String,
                        dataTypes: Seq[DataType],
                        msg: String = ""): Unit = {
    val actualDataType = schema(colName).dataType
    val message = if (msg != null && msg.trim.length > 0) " " + msg else ""
    require(dataTypes.exists(actualDataType.equals),
      s"Column $colName must be of type equal to one of the following types: " +
        s"${dataTypes.mkString("[", ", ", "]")} but was actually of type $actualDataType.$message")
  }

  var i = 0 // count the number of time the evaluate method is called
  override def evaluate(dataset: Dataset[_]): Double = {
    val schema = dataset.schema
    checkColumnTypes(schema, $(predictionCol), Seq(DoubleType, FloatType))
    checkNumericType(schema, $(labelCol))

    val predictionAndLabels = dataset
      .select(col($(predictionCol)).cast(DoubleType), col($(labelCol)).cast(DoubleType))
      .rdd
      .map { case Row(prediction: Double, label: Double) => (prediction, label) }
    val metrics = new RegressionMetrics(predictionAndLabels)
    val metric = "mae" match {
      case "rmse" => metrics.rootMeanSquaredError
      case "mse" => metrics.meanSquaredError
      case "r2" => metrics.r2
      case "mae" => metrics.meanAbsoluteError
    }
    println(s"$i $metric") // Print the metrics
    i = i + 1 // Update counter
    metric
  }

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

object RegressionEvaluator extends DefaultParamsReadable[RegressionEvaluator] {

  override def load(path: String): RegressionEvaluator = super.load(path)
}

private[ml] trait HasPredictionCol extends Params {

  /**
    * Param for prediction column name.
    * @group param
    */
  final val predictionCol: Param[String] = new Param[String](this, "predictionCol", "prediction column name")

  setDefault(predictionCol, "prediction")

  /** @group getParam */
  final def getPredictionCol: String = $(predictionCol)
}

private[ml] trait HasLabelCol extends Params {

  /**
    * Param for label column name.
    * @group param
    */
  final val labelCol: Param[String] = new Param[String](this, "labelCol", "label column name")

  setDefault(labelCol, "label")

  /** @group getParam */
  final def getLabelCol: String = $(labelCol)
}