Skip to content

[SPARK-21087] [ML] CrossValidator, TrainValidationSplit expose sub models after fitting: Scala #19208

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Closed
WeichenXu123 wants to merge 11 commits into from
Closed

[SPARK-21087] [ML] CrossValidator, TrainValidationSplit expose sub models after fitting: Scala #19208

Show file tree
Hide file tree
Changes from 5 commits
Commits
Show all changes
11 commits
Select commit Hold shift + click to select a range
46d3ab3
init pr
WeichenXu123 Sep 11, 2017
ae13440
fix style
WeichenXu123 Sep 12, 2017
e0f4ce6
remove code for dump models to disk
WeichenXu123 Sep 14, 2017
a33c4ea
merge master & resolve conflicts
WeichenXu123 Sep 27, 2017
e009ee1
address comment issues
WeichenXu123 Sep 27, 2017
931fa6c
address issues from comments
WeichenXu123 Nov 3, 2017
2a83fb5
fix style
WeichenXu123 Nov 3, 2017
f2ef609
address comments from joseph
WeichenXu123 Nov 6, 2017
7bacfca
address comments from joseph
WeichenXu123 Nov 7, 2017
654e4d5
fix mima
WeichenXu123 Nov 7, 2017
7e997da
address minor issues
WeichenXu123 Nov 14, 2017
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
6 changes: 5 additions & 1 deletion mllib/src/main/scala/org/apache/spark/ml/param/shared/SharedParamsCodeGen.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -82,7 +82,11 @@ private[shared] object SharedParamsCodeGen {
"all instance weights as 1.0"),
ParamDesc[String]("solver", "the solver algorithm for optimization", finalFields = false),
ParamDesc[Int]("aggregationDepth", "suggested depth for treeAggregate (>= 2)", Some("2"),
isValid = "ParamValidators.gtEq(2)", isExpertParam = true))
isValid = "ParamValidators.gtEq(2)", isExpertParam = true),
ParamDesc[Boolean]("collectSubModels", "whether to collect a list of sub-models trained " +
Copy link
Member

@jkbradley jkbradley Nov 1, 2017

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Some more explanation will be nice:
If set to false, then only the single best sub-model will be available after fitting.
If set to true, then all sub-models will be available. Warning: For large models, collecting all sub-models can cause OOMs on the Spark driver.

"during tuning",
Some("false"), isExpertParam = true)
)

val code = genSharedParams(params)
val file = "src/main/scala/org/apache/spark/ml/param/shared/sharedParams.scala"
Expand Down
17 changes: 17 additions & 0 deletions mllib/src/main/scala/org/apache/spark/ml/param/shared/sharedParams.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -402,4 +402,21 @@ private[ml] trait HasAggregationDepth extends Params {
/** @group expertGetParam */
final def getAggregationDepth: Int = $(aggregationDepth)
}

/**
* Trait for shared param collectSubModels (default: false).
*/
private[ml] trait HasCollectSubModels extends Params {

/**
* Param for whether to collect a list of sub-models trained during tuning.
* @group expertParam
*/
final val collectSubModels: BooleanParam = new BooleanParam(this, "collectSubModels", "whether to collect a list of sub-models trained during tuning")

setDefault(collectSubModels, false)

/** @group expertGetParam */
final def getCollectSubModels: Boolean = $(collectSubModels)
}
// scalastyle:on
107 changes: 95 additions & 12 deletions mllib/src/main/scala/org/apache/spark/ml/tuning/CrossValidator.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -17,6 +17,7 @@

package org.apache.spark.ml.tuning

import java.io.IOException
import java.util.{List => JList}

import scala.collection.JavaConverters._
Expand All @@ -31,7 +32,7 @@ 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.{IntParam, ParamMap, ParamValidators}
import org.apache.spark.ml.param.shared.HasParallelism
import org.apache.spark.ml.param.shared.{HasCollectSubModels, HasParallelism}
import org.apache.spark.ml.util._
import org.apache.spark.mllib.util.MLUtils
import org.apache.spark.sql.{DataFrame, Dataset}
Expand Down Expand Up @@ -67,7 +68,8 @@ private[ml] trait CrossValidatorParams extends ValidatorParams {
@Since("1.2.0")
class CrossValidator @Since("1.2.0") (@Since("1.4.0") override val uid: String)
extends Estimator[CrossValidatorModel]
with CrossValidatorParams with HasParallelism with MLWritable with Logging {
with CrossValidatorParams with HasParallelism with HasCollectSubModels
with MLWritable with Logging {

@Since("1.2.0")
def this() = this(Identifiable.randomUID("cv"))
Expand Down Expand Up @@ -101,6 +103,10 @@ class CrossValidator @Since("1.2.0") (@Since("1.4.0") override val uid: String)
@Since("2.3.0")
def setParallelism(value: Int): this.type = set(parallelism, value)

/** @group expertSetParam */
@Since("2.3.0")
def setCollectSubModels(value: Boolean): this.type = set(collectSubModels, value)

@Since("2.0.0")
override def fit(dataset: Dataset[_]): CrossValidatorModel = {
val schema = dataset.schema
Expand All @@ -117,6 +123,12 @@ class CrossValidator @Since("1.2.0") (@Since("1.4.0") override val uid: String)
instr.logParams(numFolds, seed, parallelism)
logTuningParams(instr)

val collectSubModelsParam = $(collectSubModels)

var subModels: Option[Array[Array[Model[_]]]] = if (collectSubModelsParam) {
Copy link
Contributor

@holdenk holdenk Nov 2, 2017

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

so this var seems unnecessary, could we just it seems like we'd be better by just collecting modelFutures in copy values (then we can avoid the mutation on L145)

Copy link
Contributor Author

@WeichenXu123 WeichenXu123 Nov 3, 2017 •
edited
Loading

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

@holdenk @jkbradley I already thought about this issue. The reason I use this way is:

  1. The modelFutures and foldMetricFutures will be executed in pipelined way, when $(collectSubModels) == false, this will make sure that the model generated in modelFutures will be released in time, so that the maximum memory cost will be numParallelism * sizeof(model). If we use the way of "collecting modelFutures", it will increase the memory cost to be $(estimatorParamMaps).length * sizeof(model) . This is a serious issue which is discussed before.
  2. IMO the mutation on L145 won't influence performance. and it do not need something like lock, there is no race condition.

Copy link
Contributor

@holdenk holdenk Nov 4, 2017

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I don't follow with #1, if we keep all the models (e.g. set collectSubModelsParam) then the maximum memory cost will be $(estimatorParamMaps).length * sizeof(model) in either case? If we don't keep the models (e.g. set collectSubModelsParam to false) then you don't have to collect the future back at the end and there is no additional overhead.

For #2, It's not that mutation impacts performance, its that it makes the code less easy to reason about for no gain (unless I've misunderstood something about part 1).

Copy link
Contributor Author

@WeichenXu123 WeichenXu123 Nov 4, 2017 •
edited
Loading

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

@holdenk Oh, sorry for confusing you. Yes, if set collectSubModelsParam the memory cost will always be $(estimatorParamMaps).length * sizeof(model). According to your suggestion, we have to duplicate code logic (but if i am wrong correct me):

  • When set collectSubModelsParam, we cannot pipeline modelFutures and foldMetricFutures, we should execute modelFutures and collect results first, and modify foldMetricFutures logic, change it into something like following:
val foldMetricFutures = modelResults.zip(epm).map { case (model, paramMap) =>
       Future[Double] {
          val metric = eval.evaluate(model.transform(validationDataset, paramMap))
          logDebug(s"Got metric $metric for model trained with $paramMap.")
          metric
      } (executionContext)
  • When not set collectSubModelsParam, just keep current modelFutures & foldMetricFutures and pipeline them to execute. (Only pipeline them we can save memory cost to numParallelism * sizeof(model).

So, according to your suggestion, it seems need more code. So do you still prefer this way ? Or do you have better way to implement that ?

Copy link
Member

@jkbradley jkbradley Nov 11, 2017

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Sorry I didn't follow up on this before. I think that @WeichenXu123 's argument is valid, but please say if there are issues I'm missing @holdenk

Some(Array.fill($(numFolds))(Array.fill[Model[_]](epm.length)(null)))
} else None

// Compute metrics for each model over each split
val splits = MLUtils.kFold(dataset.toDF.rdd, $(numFolds), $(seed))
val metrics = splits.zipWithIndex.map { case ((training, validation), splitIndex) =>
Expand All @@ -125,10 +137,14 @@ class CrossValidator @Since("1.2.0") (@Since("1.4.0") override val uid: String)
logDebug(s"Train split $splitIndex with multiple sets of parameters.")

// Fit models in a Future for training in parallel
val modelFutures = epm.map { paramMap =>
val modelFutures = epm.zipWithIndex.map { case (paramMap, paramIndex) =>
Future[Model[_]] {
val model = est.fit(trainingDataset, paramMap)
model.asInstanceOf[Model[_]]
val model = est.fit(trainingDataset, paramMap).asInstanceOf[Model[_]]

if (collectSubModelsParam) {
subModels.get(splitIndex)(paramIndex) = model
}
model
} (executionContext)
}

Expand Down Expand Up @@ -160,7 +176,7 @@ 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))
copyValues(new CrossValidatorModel(uid, bestModel, metrics, subModels).setParent(this))
}

@Since("1.4.0")
Expand Down Expand Up @@ -236,12 +252,17 @@ object CrossValidator extends MLReadable[CrossValidator] {
class CrossValidatorModel private[ml] (
@Since("1.4.0") override val uid: String,
@Since("1.2.0") val bestModel: Model[_],
@Since("1.5.0") val avgMetrics: Array[Double])
@Since("1.5.0") val avgMetrics: Array[Double],
@Since("2.3.0") val subModels: Option[Array[Array[Model[_]]]])
Copy link
Member

@jkbradley jkbradley Nov 1, 2017

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

This API (Option) won't be Java-friendly. I'd prefer to follow this pattern, which has been used for optional fields such as model summaries:

  • private val which is an Option
  • public method subModels: Array[Array[Model[_]]] which throws an Exception if the sub-models are not available
  • public method hasSubModels: Boolean for checking if subModels is available

Copy link
Contributor Author

@WeichenXu123 WeichenXu123 Nov 3, 2017

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

So also need a setSubModels method for setting it.

extends Model[CrossValidatorModel] with CrossValidatorParams with MLWritable {

/** A Python-friendly auxiliary constructor. */
private[ml] def this(uid: String, bestModel: Model[_], avgMetrics: JList[Double]) = {
this(uid, bestModel, avgMetrics.asScala.toArray)
this(uid, bestModel, avgMetrics.asScala.toArray, null)
Copy link
Contributor

@smurching smurching Sep 18, 2017

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

See earlier suggestion, use an Option set to None instead of setting the Array to null

}

private[ml] def this(uid: String, bestModel: Model[_], avgMetrics: Array[Double]) = {
this(uid, bestModel, avgMetrics, null)
}

@Since("2.0.0")
Expand All @@ -260,17 +281,39 @@ class CrossValidatorModel private[ml] (
val copied = new CrossValidatorModel(
uid,
bestModel.copy(extra).asInstanceOf[Model[_]],
avgMetrics.clone())
avgMetrics.clone(),
CrossValidatorModel.copySubModels(subModels))
copyValues(copied, extra).setParent(parent)
}

@Since("1.6.0")
override def write: MLWriter = new CrossValidatorModel.CrossValidatorModelWriter(this)

@Since("2.3.0")
@throws[IOException]("If the input path already exists but overwrite is not enabled.")
def save(path: String, persistSubModels: Boolean): Unit = {
write.asInstanceOf[CrossValidatorModel.CrossValidatorModelWriter]
.persistSubModels(persistSubModels).save(path)
}
Copy link
Contributor Author

@WeichenXu123 WeichenXu123 Sep 12, 2017

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I add this method because the CrossValidatorModelWriter is private. User cannot use it. But I don't know whether there is better solution.

Copy link
Contributor

@smurching smurching Sep 19, 2017

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I think users can still access CrossValidatorModelWriter through CrossValidatorModel.write, so the save method is unnecessary.

The private[CrossValidatorModel] annotation on the CrossValidatorModelWriter constructor only means that users can't create instances of the class e.g. via new CrossValidatorModel.CrossValidatorModelWriter(...)

Copy link
Contributor Author

@WeichenXu123 WeichenXu123 Sep 27, 2017

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I tried model.write.asInstanceOf[CrossValidatorModel.CrossValidatorModelWriter] but cannot pass complier, it is inaccessible.
Do you have some other ways ?

Copy link
Contributor Author

@WeichenXu123 WeichenXu123 Sep 27, 2017

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Discussion: Another way I think is adding an interface def option(key: String, value: String) into Writer. cc @jkbradley

Copy link
Member

@jkbradley jkbradley Nov 3, 2017

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I agree with the last suggestion of adding def option(key: String, value: String) to mimic the SQL datasource API.

}

@Since("1.6.0")
object CrossValidatorModel extends MLReadable[CrossValidatorModel] {

private[CrossValidatorModel] def copySubModels(subModels: Option[Array[Array[Model[_]]]]) = {
Copy link
Member

@jkbradley jkbradley Nov 3, 2017

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

style: state return value explicitly

subModels.map { subModels =>
Copy link
Member

@jkbradley jkbradley Nov 3, 2017

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Can this be simplified using map?

subModels.map(_.map(_.map(_.copy(...).asInstanceOf[...])))

val numFolds = subModels.length
val numParamMaps = subModels(0).length
val copiedSubModels = Array.fill(numFolds)(Array.fill[Model[_]](numParamMaps)(null))
for (i <- 0 until numFolds) {
for (j <- 0 until numParamMaps) {
copiedSubModels(i)(j) = subModels(i)(j).copy(ParamMap.empty).asInstanceOf[Model[_]]
}
}
copiedSubModels
}
}

@Since("1.6.0")
override def read: MLReader[CrossValidatorModel] = new CrossValidatorModelReader

Expand All @@ -282,12 +325,35 @@ object CrossValidatorModel extends MLReadable[CrossValidatorModel] {

ValidatorParams.validateParams(instance)

protected var shouldPersistSubModels: Boolean = false

/**
* Set option for persist sub models.
*/
@Since("2.3.0")
def persistSubModels(persist: Boolean): this.type = {
shouldPersistSubModels = persist
this
}

override protected def saveImpl(path: String): Unit = {
import org.json4s.JsonDSL._
val extraMetadata = "avgMetrics" -> instance.avgMetrics.toSeq
val extraMetadata = ("avgMetrics" -> instance.avgMetrics.toSeq) ~
("shouldPersistSubModels" -> shouldPersistSubModels)
Copy link
Member

@jkbradley jkbradley Nov 3, 2017

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Let's have 1 name for this argument: "persistSubModels"

ValidatorParams.saveImpl(path, instance, sc, Some(extraMetadata))
val bestModelPath = new Path(path, "bestModel").toString
instance.bestModel.asInstanceOf[MLWritable].save(bestModelPath)
if (shouldPersistSubModels) {
require(instance.subModels.isDefined, "Cannot get sub models to persist.")
val subModelsPath = new Path(path, "subModels")
for (splitIndex <- 0 until instance.getNumFolds) {
val splitPath = new Path(subModelsPath, splitIndex.toString)
Copy link
Member

@jkbradley jkbradley Nov 3, 2017

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

How about naming this with the string "fold":
splitIndex.toString --> "fold" + splitIndex.toString?

for (paramIndex <- 0 until instance.getEstimatorParamMaps.length) {
val modelPath = new Path(splitPath, paramIndex.toString).toString
instance.subModels.get(splitIndex)(paramIndex).asInstanceOf[MLWritable].save(modelPath)
}
}
}
}
}

Expand All @@ -301,11 +367,28 @@ object CrossValidatorModel extends MLReadable[CrossValidatorModel] {

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

val model = new CrossValidatorModel(metadata.uid, bestModel, avgMetrics)
val shouldPersistSubModels = (metadata.metadata \ "shouldPersistSubModels").extract[Boolean]

val subModels: Option[Array[Array[Model[_]]]] = if (shouldPersistSubModels) {
val subModelsPath = new Path(path, "subModels")
val _subModels = Array.fill(numFolds)(Array.fill[Model[_]](
estimatorParamMaps.length)(null))
for (splitIndex <- 0 until numFolds) {
val splitPath = new Path(subModelsPath, splitIndex.toString)
for (paramIndex <- 0 until estimatorParamMaps.length) {
val modelPath = new Path(splitPath, paramIndex.toString).toString
_subModels(splitIndex)(paramIndex) =
DefaultParamsReader.loadParamsInstance(modelPath, sc)
}
}
Some(_subModels)
} else None

val model = new CrossValidatorModel(metadata.uid, bestModel, avgMetrics, subModels)
model.set(model.estimator, estimator)
.set(model.evaluator, evaluator)
.set(model.estimatorParamMaps, estimatorParamMaps)
Expand Down
Loading