[SPARK-11893] Model export/import for spark.ml: TrainValidationSplit

https://issues.apache.org/jira/browse/SPARK-11893

jkbradley In order to share read/write with `TrainValidationSplit`, I move the `SharedReadWrite` out of `CrossValidator` into a new trait `SharedReadWrite` in the tunning package.

To reduce the repeated tests, I move the complex tests from `CrossValidatorSuite` to `SharedReadWriteSuite`, and create a fake validator called `MyValidator` to test the shared code.

With `SharedReadWrite`, potential newly added `Validator` can share the read/write common part, and only need to implement their extra params save/load.

Author: Xusen Yin <[email protected]>
Author: Joseph K. Bradley <[email protected]>

Closes #9971 from yinxusen/SPARK-11893.

Author: yinxusen

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

@@ -19,32 +19,27 @@ package org.apache.spark.ml.tuning
 
 import com.github.fommil.netlib.F2jBLAS
 import org.apache.hadoop.fs.Path
-import org.json4s.{DefaultFormats, JObject}
-import org.json4s.jackson.JsonMethods._
+import org.json4s.DefaultFormats
 
-import org.apache.spark.SparkContext
 import org.apache.spark.annotation.{Experimental, Since}
 import org.apache.spark.internal.Logging
 import org.apache.spark.ml._
-import org.apache.spark.ml.classification.OneVsRestParams
 import org.apache.spark.ml.evaluation.Evaluator
-import org.apache.spark.ml.feature.RFormulaModel
 import org.apache.spark.ml.param._
 import org.apache.spark.ml.param.shared.HasSeed
 import org.apache.spark.ml.util._
-import org.apache.spark.ml.util.DefaultParamsReader.Metadata
 import org.apache.spark.mllib.util.MLUtils
 import org.apache.spark.sql.DataFrame
 import org.apache.spark.sql.types.StructType
 
-
 /**
  * Params for [[CrossValidator]] and [[CrossValidatorModel]].
  */
 private[ml] trait CrossValidatorParams extends ValidatorParams with HasSeed {
   /**
    * Param for number of folds for cross validation.  Must be >= 2.
    * Default: 3
+   *
    * @group param
    */
   val numFolds: IntParam = new IntParam(this, "numFolds",
@@ -163,10 +158,10 @@ object CrossValidator extends MLReadable[CrossValidator] {
 
   private[CrossValidator] class CrossValidatorWriter(instance: CrossValidator) extends MLWriter {
 
-    SharedReadWrite.validateParams(instance)
+    ValidatorParams.validateParams(instance)
 
     override protected def saveImpl(path: String): Unit =
-      SharedReadWrite.saveImpl(path, instance, sc)
+      ValidatorParams.saveImpl(path, instance, sc)
   }
 
   private class CrossValidatorReader extends MLReader[CrossValidator] {
@@ -175,132 +170,18 @@ object CrossValidator extends MLReadable[CrossValidator] {
     private val className = classOf[CrossValidator].getName
 
     override def load(path: String): CrossValidator = {
-      val (metadata, estimator, evaluator, estimatorParamMaps, numFolds) =
-        SharedReadWrite.load(path, sc, className)
+      implicit val format = DefaultFormats
+
+      val (metadata, estimator, evaluator, estimatorParamMaps) =
+        ValidatorParams.loadImpl(path, sc, className)
+      val numFolds = (metadata.params \ "numFolds").extract[Int]
       new CrossValidator(metadata.uid)
         .setEstimator(estimator)
         .setEvaluator(evaluator)
         .setEstimatorParamMaps(estimatorParamMaps)
         .setNumFolds(numFolds)
     }
   }
-
-  private object CrossValidatorReader {
-    /**
-     * Examine the given estimator (which may be a compound estimator) and extract a mapping
-     * from UIDs to corresponding [[Params]] instances.
-     */
-    def getUidMap(instance: Params): Map[String, Params] = {
-      val uidList = getUidMapImpl(instance)
-      val uidMap = uidList.toMap
-      if (uidList.size != uidMap.size) {
-        throw new RuntimeException("CrossValidator.load found a compound estimator with stages" +
-          s" with duplicate UIDs.  List of UIDs: ${uidList.map(_._1).mkString(", ")}")
-      }
-      uidMap
-    }
-
-    def getUidMapImpl(instance: Params): List[(String, Params)] = {
-      val subStages: Array[Params] = instance match {
-        case p: Pipeline => p.getStages.asInstanceOf[Array[Params]]
-        case pm: PipelineModel => pm.stages.asInstanceOf[Array[Params]]
-        case v: ValidatorParams => Array(v.getEstimator, v.getEvaluator)
-        case ovr: OneVsRestParams =>
-          // TODO: SPARK-11892: This case may require special handling.
-          throw new UnsupportedOperationException("CrossValidator write will fail because it" +
-            " cannot yet handle an estimator containing type: ${ovr.getClass.getName}")
-        case rformModel: RFormulaModel => Array(rformModel.pipelineModel)
-        case _: Params => Array()
-      }
-      val subStageMaps = subStages.map(getUidMapImpl).foldLeft(List.empty[(String, Params)])(_ ++ _)
-      List((instance.uid, instance)) ++ subStageMaps
-    }
-  }
-
-  private[tuning] object SharedReadWrite {
-
-    /**
-     * Check that [[CrossValidator.evaluator]] and [[CrossValidator.estimator]] are Writable.
-     * This does not check [[CrossValidator.estimatorParamMaps]].
-     */
-    def validateParams(instance: ValidatorParams): Unit = {
-      def checkElement(elem: Params, name: String): Unit = elem match {
-        case stage: MLWritable => // good
-        case other =>
-          throw new UnsupportedOperationException("CrossValidator write will fail " +
-            s" because it contains $name which does not implement Writable." +
-            s" Non-Writable $name: ${other.uid} of type ${other.getClass}")
-      }
-      checkElement(instance.getEvaluator, "evaluator")
-      checkElement(instance.getEstimator, "estimator")
-      // Check to make sure all Params apply to this estimator.  Throw an error if any do not.
-      // Extraneous Params would cause problems when loading the estimatorParamMaps.
-      val uidToInstance: Map[String, Params] = CrossValidatorReader.getUidMap(instance)
-      instance.getEstimatorParamMaps.foreach { case pMap: ParamMap =>
-        pMap.toSeq.foreach { case ParamPair(p, v) =>
-          require(uidToInstance.contains(p.parent), s"CrossValidator save requires all Params in" +
-            s" estimatorParamMaps to apply to this CrossValidator, its Estimator, or its" +
-            s" Evaluator.  An extraneous Param was found: $p")
-        }
-      }
-    }
-
-    private[tuning] def saveImpl(
-        path: String,
-        instance: CrossValidatorParams,
-        sc: SparkContext,
-        extraMetadata: Option[JObject] = None): Unit = {
-      import org.json4s.JsonDSL._
-
-      val estimatorParamMapsJson = compact(render(
-        instance.getEstimatorParamMaps.map { case paramMap =>
-          paramMap.toSeq.map { case ParamPair(p, v) =>
-            Map("parent" -> p.parent, "name" -> p.name, "value" -> p.jsonEncode(v))
-          }
-        }.toSeq
-      ))
-      val jsonParams = List(
-        "numFolds" -> parse(instance.numFolds.jsonEncode(instance.getNumFolds)),
-        "estimatorParamMaps" -> parse(estimatorParamMapsJson)
-      )
-      DefaultParamsWriter.saveMetadata(instance, path, sc, extraMetadata, Some(jsonParams))
-
-      val evaluatorPath = new Path(path, "evaluator").toString
-      instance.getEvaluator.asInstanceOf[MLWritable].save(evaluatorPath)
-      val estimatorPath = new Path(path, "estimator").toString
-      instance.getEstimator.asInstanceOf[MLWritable].save(estimatorPath)
-    }
-
-    private[tuning] def load[M <: Model[M]](
-        path: String,
-        sc: SparkContext,
-        expectedClassName: String): (Metadata, Estimator[M], Evaluator, Array[ParamMap], Int) = {
-
-      val metadata = DefaultParamsReader.loadMetadata(path, sc, expectedClassName)
-
-      implicit val format = DefaultFormats
-      val evaluatorPath = new Path(path, "evaluator").toString
-      val evaluator = DefaultParamsReader.loadParamsInstance[Evaluator](evaluatorPath, sc)
-      val estimatorPath = new Path(path, "estimator").toString
-      val estimator = DefaultParamsReader.loadParamsInstance[Estimator[M]](estimatorPath, sc)
-
-      val uidToParams = Map(evaluator.uid -> evaluator) ++ CrossValidatorReader.getUidMap(estimator)
-
-      val numFolds = (metadata.params \ "numFolds").extract[Int]
-      val estimatorParamMaps: Array[ParamMap] =
-        (metadata.params \ "estimatorParamMaps").extract[Seq[Seq[Map[String, String]]]].map {
-          pMap =>
-            val paramPairs = pMap.map { case pInfo: Map[String, String] =>
-              val est = uidToParams(pInfo("parent"))
-              val param = est.getParam(pInfo("name"))
-              val value = param.jsonDecode(pInfo("value"))
-              param -> value
-            }
-            ParamMap(paramPairs: _*)
-        }.toArray
-      (metadata, estimator, evaluator, estimatorParamMaps, numFolds)
-    }
-  }
 }
 
 /**
@@ -346,8 +227,6 @@ class CrossValidatorModel private[ml] (
 @Since("1.6.0")
 object CrossValidatorModel extends MLReadable[CrossValidatorModel] {
 
-  import CrossValidator.SharedReadWrite
-
   @Since("1.6.0")
   override def read: MLReader[CrossValidatorModel] = new CrossValidatorModelReader
 
@@ -357,12 +236,12 @@ object CrossValidatorModel extends MLReadable[CrossValidatorModel] {
   private[CrossValidatorModel]
   class CrossValidatorModelWriter(instance: CrossValidatorModel) extends MLWriter {
 
-    SharedReadWrite.validateParams(instance)
+    ValidatorParams.validateParams(instance)
 
     override protected def saveImpl(path: String): Unit = {
       import org.json4s.JsonDSL._
       val extraMetadata = "avgMetrics" -> instance.avgMetrics.toSeq
-      SharedReadWrite.saveImpl(path, instance, sc, Some(extraMetadata))
+      ValidatorParams.saveImpl(path, instance, sc, Some(extraMetadata))
       val bestModelPath = new Path(path, "bestModel").toString
       instance.bestModel.asInstanceOf[MLWritable].save(bestModelPath)
     }
@@ -376,8 +255,9 @@ object CrossValidatorModel extends MLReadable[CrossValidatorModel] {
     override def load(path: String): CrossValidatorModel = {
       implicit val format = DefaultFormats
 
-      val (metadata, estimator, evaluator, estimatorParamMaps, numFolds) =
-        SharedReadWrite.load(path, sc, className)
+      val (metadata, estimator, evaluator, estimatorParamMaps) =
+        ValidatorParams.loadImpl(path, sc, className)
+      val numFolds = (metadata.params \ "numFolds").extract[Int]
       val bestModelPath = new Path(path, "bestModel").toString
       val bestModel = DefaultParamsReader.loadParamsInstance[Model[_]](bestModelPath, sc)
       val avgMetrics = (metadata.metadata \ "avgMetrics").extract[Seq[Double]].toArray

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

@@ -17,12 +17,15 @@
 
 package org.apache.spark.ml.tuning
 
+import org.apache.hadoop.fs.Path
+import org.json4s.DefaultFormats
+
 import org.apache.spark.annotation.{Experimental, Since}
 import org.apache.spark.internal.Logging
 import org.apache.spark.ml.{Estimator, Model}
 import org.apache.spark.ml.evaluation.Evaluator
 import org.apache.spark.ml.param.{DoubleParam, ParamMap, ParamValidators}
-import org.apache.spark.ml.util.Identifiable
+import org.apache.spark.ml.util._
 import org.apache.spark.sql.DataFrame
 import org.apache.spark.sql.types.StructType
 
@@ -33,6 +36,7 @@ private[ml] trait TrainValidationSplitParams extends ValidatorParams {
   /**
    * Param for ratio between train and validation data. Must be between 0 and 1.
    * Default: 0.75
+   *
    * @group param
    */
   val trainRatio: DoubleParam = new DoubleParam(this, "trainRatio",
@@ -55,7 +59,7 @@ private[ml] trait TrainValidationSplitParams extends ValidatorParams {
 @Experimental
 class TrainValidationSplit @Since("1.5.0") (@Since("1.5.0") override val uid: String)
   extends Estimator[TrainValidationSplitModel]
-  with TrainValidationSplitParams with Logging {
+  with TrainValidationSplitParams with MLWritable with Logging {
 
   @Since("1.5.0")
   def this() = this(Identifiable.randomUID("tvs"))
@@ -130,6 +134,47 @@ class TrainValidationSplit @Since("1.5.0") (@Since("1.5.0") override val uid: St
     }
     copied
   }
+
+  @Since("2.0.0")
+  override def write: MLWriter = new TrainValidationSplit.TrainValidationSplitWriter(this)
+}
+
+@Since("2.0.0")
+object TrainValidationSplit extends MLReadable[TrainValidationSplit] {
+
+  @Since("2.0.0")
+  override def read: MLReader[TrainValidationSplit] = new TrainValidationSplitReader
+
+  @Since("2.0.0")
+  override def load(path: String): TrainValidationSplit = super.load(path)
+
+  private[TrainValidationSplit] class TrainValidationSplitWriter(instance: TrainValidationSplit)
+    extends MLWriter {
+
+    ValidatorParams.validateParams(instance)
+
+    override protected def saveImpl(path: String): Unit =
+      ValidatorParams.saveImpl(path, instance, sc)
+  }
+
+  private class TrainValidationSplitReader extends MLReader[TrainValidationSplit] {
+
+    /** Checked against metadata when loading model */
+    private val className = classOf[TrainValidationSplit].getName
+
+    override def load(path: String): TrainValidationSplit = {
+      implicit val format = DefaultFormats
+
+      val (metadata, estimator, evaluator, estimatorParamMaps) =
+        ValidatorParams.loadImpl(path, sc, className)
+      val trainRatio = (metadata.params \ "trainRatio").extract[Double]
+      new TrainValidationSplit(metadata.uid)
+        .setEstimator(estimator)
+        .setEvaluator(evaluator)
+        .setEstimatorParamMaps(estimatorParamMaps)
+        .setTrainRatio(trainRatio)
+    }
+  }
 }
 
 /**
@@ -146,7 +191,7 @@ class TrainValidationSplitModel private[ml] (
     @Since("1.5.0") override val uid: String,
     @Since("1.5.0") val bestModel: Model[_],
     @Since("1.5.0") val validationMetrics: Array[Double])
-  extends Model[TrainValidationSplitModel] with TrainValidationSplitParams {
+  extends Model[TrainValidationSplitModel] with TrainValidationSplitParams with MLWritable {
 
   @Since("1.5.0")
   override def transform(dataset: DataFrame): DataFrame = {
@@ -167,4 +212,53 @@ class TrainValidationSplitModel private[ml] (
       validationMetrics.clone())
     copyValues(copied, extra)
   }
+
+  @Since("2.0.0")
+  override def write: MLWriter = new TrainValidationSplitModel.TrainValidationSplitModelWriter(this)
+}
+
+@Since("2.0.0")
+object TrainValidationSplitModel extends MLReadable[TrainValidationSplitModel] {
+
+  @Since("2.0.0")
+  override def read: MLReader[TrainValidationSplitModel] = new TrainValidationSplitModelReader
+
+  @Since("2.0.0")
+  override def load(path: String): TrainValidationSplitModel = super.load(path)
+
+  private[TrainValidationSplitModel]
+  class TrainValidationSplitModelWriter(instance: TrainValidationSplitModel) extends MLWriter {
+
+    ValidatorParams.validateParams(instance)
+
+    override protected def saveImpl(path: String): Unit = {
+      import org.json4s.JsonDSL._
+      val extraMetadata = "validationMetrics" -> instance.validationMetrics.toSeq
+      ValidatorParams.saveImpl(path, instance, sc, Some(extraMetadata))
+      val bestModelPath = new Path(path, "bestModel").toString
+      instance.bestModel.asInstanceOf[MLWritable].save(bestModelPath)
+    }
+  }
+
+  private class TrainValidationSplitModelReader extends MLReader[TrainValidationSplitModel] {
+
+    /** Checked against metadata when loading model */
+    private val className = classOf[TrainValidationSplitModel].getName
+
+    override def load(path: String): TrainValidationSplitModel = {
+      implicit val format = DefaultFormats
+
+      val (metadata, estimator, evaluator, estimatorParamMaps) =
+        ValidatorParams.loadImpl(path, sc, className)
+      val trainRatio = (metadata.params \ "trainRatio").extract[Double]
+      val bestModelPath = new Path(path, "bestModel").toString
+      val bestModel = DefaultParamsReader.loadParamsInstance[Model[_]](bestModelPath, sc)
+      val validationMetrics = (metadata.metadata \ "validationMetrics").extract[Seq[Double]].toArray
+      val tvs = new TrainValidationSplitModel(metadata.uid, bestModel, validationMetrics)
+      tvs.set(tvs.estimator, estimator)
+        .set(tvs.evaluator, evaluator)
+        .set(tvs.estimatorParamMaps, estimatorParamMaps)
+        .set(tvs.trainRatio, trainRatio)
+    }
+  }
 }