support cross validation

Author: YY-OnCall

examples/src/main/scala/org/apache/spark/examples/ml/ModelSelectionViaCrossValidationExample.scala

@@ -112,6 +112,7 @@ object ModelSelectionViaCrossValidationExample {
       .foreach { case Row(id: Long, text: String, prob: Vector, prediction: Double) =>
         println(s"($id, $text) --> prob=$prob, prediction=$prediction")
       }
+    cvModel.summary.trainingMetrics.show()
     // $example off$
 
     spark.stop()

examples/src/main/scala/org/apache/spark/examples/ml/ModelSelectionViaTrainValidationSplitExample.scala

@@ -74,6 +74,7 @@ object ModelSelectionViaTrainValidationSplitExample {
     model.transform(test)
       .select("features", "label", "prediction")
       .show()
+    model.summary.trainingMetrics.show()
     // $example off$
 
     spark.stop()

mllib/src/main/scala/org/apache/spark/ml/tuning/CrossValidator.scala

@@ -25,6 +25,7 @@ import com.github.fommil.netlib.F2jBLAS
 import org.apache.hadoop.fs.Path
 import org.json4s.DefaultFormats
 
+import org.apache.spark.SparkException
 import org.apache.spark.annotation.Since
 import org.apache.spark.internal.Logging
 import org.apache.spark.ml._
@@ -133,7 +134,10 @@ class CrossValidator @Since("1.2.0") (@Since("1.4.0") override val uid: String)
     logInfo(s"Best cross-validation metric: $bestMetric.")
     val bestModel = est.fit(dataset, epm(bestIndex)).asInstanceOf[Model[_]]
     instr.logSuccess(bestModel)
-    copyValues(new CrossValidatorModel(uid, bestModel, metrics).setParent(this))
+    val model = new CrossValidatorModel(uid, bestModel, metrics).setParent(this)
+    val summary = new TuningSummary(epm, metrics, bestIndex)
+    model.setSummary(Some(summary))
+    copyValues(model)
   }
 
   @Since("1.4.0")
@@ -229,6 +233,29 @@ class CrossValidatorModel private[ml] (
     bestModel.transformSchema(schema)
   }
 
+  private var trainingSummary: Option[TuningSummary] = None
+
+  private[tuning] def setSummary(summary: Option[TuningSummary]): this.type = {
+    this.trainingSummary = summary
+    this
+  }
+
+  /**
+   * Return true if there exists summary of model.
+   */
+  @Since("2.3.0")
+  def hasSummary: Boolean = trainingSummary.nonEmpty
+
+  /**
+   * Gets summary of model on training set. An exception is
+   * thrown if `trainingSummary == None`.
+   */
+  @Since("2.3.0")
+  def summary: TuningSummary = trainingSummary.getOrElse {
+    throw new SparkException(
+      s"No training summary available for the ${this.getClass.getSimpleName}")
+  }
+
   @Since("1.4.0")
   override def copy(extra: ParamMap): CrossValidatorModel = {
     val copied = new CrossValidatorModel(

mllib/src/main/scala/org/apache/spark/ml/tuning/TuningSummary.scala

@@ -14,7 +14,6 @@
  * See the License for the specific language governing permissions and
  * limitations under the License.
  */
-
 package org.apache.spark.ml.tuning
 
 import org.apache.spark.annotation.{Experimental, Since}
@@ -29,15 +28,16 @@ import org.apache.spark.sql.types.{StringType, StructField, StructType}
  * @param params  estimator param maps
  * @param metrics  Corresponding evaluation metrics for the param maps
  */
-@Since("2.2.0")
+@Since("2.3.0")
 @Experimental
-class TuningSummary private[tuning](
+private[tuning] class TuningSummary private[tuning](
     val params: Array[ParamMap],
     val metrics: Array[Double],
     val bestIndex: Int) {
 
   /**
-   * Summary of grid search tuning in the format of DataFrame.
+   * Summary of grid search tuning in the format of DataFrame. Each row contains one candidate
+   * paramMap and its corresponding metrics.
    */
   def trainingMetrics: DataFrame = {
     require(params.nonEmpty, "estimator param maps should not be empty")

mllib/src/test/scala/org/apache/spark/ml/tuning/CrossValidatorSuite.scala

@@ -66,6 +66,34 @@ class CrossValidatorSuite
     assert(cvModel.avgMetrics.length === lrParamMaps.length)
   }
 
+  test("cross validation with tuning summary") {
+    val lr = new LogisticRegression
+    val lrParamMaps = new ParamGridBuilder()
+      .addGrid(lr.regParam, Array(0.001, 1.0, 1000.0))
+      .addGrid(lr.maxIter, Array(0, 2))
+      .build()
+    val eval = new BinaryClassificationEvaluator
+    val cv = new CrossValidator()
+      .setEstimator(lr)
+      .setEstimatorParamMaps(lrParamMaps)
+      .setEvaluator(eval)
+      .setNumFolds(3)
+    val cvModel = cv.fit(dataset)
+    assert(cvModel.hasSummary)
+    assert(cvModel.summary.params === lrParamMaps)
+    assert(cvModel.summary.trainingMetrics.count() === lrParamMaps.length)
+    val expectedSummary = spark.createDataFrame(Seq(
+      (0, 0.001),
+      (2, 0.001),
+      (0, 1.0),
+      (2, 1.0),
+      (0, 1000.0),
+      (2, 1000.0)
+    ).map(t => (t._1.toString, t._2.toString))).toDF("maxIter", "regParam")
+    assert(cvModel.summary.trainingMetrics.select("maxIter", "regParam").collect().toSet
+      .equals(expectedSummary.collect().toSet))
+  }
+
   test("cross validation with linear regression") {
     val dataset = sc.parallelize(
       LinearDataGenerator.generateLinearInput(

mllib/src/test/scala/org/apache/spark/ml/tuning/TrainValidationSplitSuite.scala

@@ -51,15 +51,40 @@ class TrainValidationSplitSuite
       .setEvaluator(eval)
       .setTrainRatio(0.5)
       .setSeed(42L)
-    val tvModel = tvs.fit(dataset)
-    val parent = tvModel.bestModel.parent.asInstanceOf[LogisticRegression]
+    val tvsModel = tvs.fit(dataset)
+    val parent = tvsModel.bestModel.parent.asInstanceOf[LogisticRegression]
     assert(tvs.getTrainRatio === 0.5)
     assert(parent.getRegParam === 0.001)
     assert(parent.getMaxIter === 10)
-    assert(tvModel.validationMetrics.length === lrParamMaps.length)
-    assert(tvModel.summary.params === lrParamMaps)
-    assert(tvModel.summary.trainingMetrics.count() === lrParamMaps.length)
-    assert(tvModel.summary.trainingMetrics.columns === Array("maxIter", "regParam", "metrics"))
+    assert(tvsModel.validationMetrics.length === lrParamMaps.length)
+  }
+
+  test("train validation split with tuning summary") {
+    val dataset = sc.parallelize(generateLogisticInput(1.0, 1.0, 100, 42), 2).toDF()
+    val lr = new LogisticRegression
+    val lrParamMaps = new ParamGridBuilder()
+      .addGrid(lr.regParam, Array(0.001, 1.0, 1000.0))
+      .addGrid(lr.maxIter, Array(0, 2))
+      .build()
+    val eval = new BinaryClassificationEvaluator
+    val tvs = new TrainValidationSplit()
+      .setEstimator(lr)
+      .setEstimatorParamMaps(lrParamMaps)
+      .setEvaluator(eval)
+    val tvsModel = tvs.fit(dataset)
+    assert(tvsModel.hasSummary)
+    assert(tvsModel.summary.params === lrParamMaps)
+    assert(tvsModel.summary.trainingMetrics.count() === lrParamMaps.length)
+    val expectedSummary = spark.createDataFrame(Seq(
+      (0, 0.001),
+      (2, 0.001),
+      (0, 1.0),
+      (2, 1.0),
+      (0, 1000.0),
+      (2, 1000.0)
+    ).map(t => (t._1.toString, t._2.toString))).toDF("maxIter", "regParam")
+    assert(tvsModel.summary.trainingMetrics.select("maxIter", "regParam").collect().toSet
+      .equals(expectedSummary.collect().toSet))
   }
 
   test("train validation with linear regression") {
@@ -89,7 +114,7 @@ class TrainValidationSplitSuite
     assert(parent.getMaxIter === 10)
     assert(tvsModel.validationMetrics.length === lrParamMaps.length)
 
-      eval.setMetricName("r2")
+    eval.setMetricName("r2")
     val tvsModel2 = tvs.fit(dataset)
     val parent2 = tvsModel2.bestModel.parent.asInstanceOf[LinearRegression]
     assert(parent2.getRegParam === 0.001)