(spark) branch branch-4.x updated: [SPARK-58462][ML][CONNECT] Exclude tuning params from CrossValidatorModel size

[ru...@apache.org]

This is an automated email from the ASF dual-hosted git repository.

zhengruifeng pushed a commit to branch branch-4.x
in repository https://gitbox.apache.org/repos/asf/spark.git


The following commit(s) were added to refs/heads/branch-4.x by this push:
     new b2142d238703 [SPARK-58462][ML][CONNECT] Exclude tuning params from CrossValidatorModel size
b2142d238703 is described below

commit b2142d23870395ad9a15b80d5ea1228fd9e68ab6
Author: Ruifeng Zheng <ru...@apache.org>
AuthorDate: Fri Jul 31 20:34:34 2026 +0800

    [SPARK-58462][ML][CONNECT] Exclude tuning params from CrossValidatorModel size
    
    ### What changes were proposed in this pull request?
    
    This patch adds an exclusion list to `Params.estimateMatadataSize` and uses it from `CrossValidatorModel.estimatedSize` to omit the estimator, evaluator, and estimator parameter maps. The model estimate instead includes the learned best model, collected submodels, metrics, and remaining metadata.
    
    ### Why are the changes needed?
    
    CrossValidatorModel retains tuning configuration in its parameter maps. Estimators and their parameter validators can retain logger or runtime state that reaches SparkSession and SparkContext. That shared state is not learned model data and can inflate ML Connect cache accounting.
    
    ### Does this PR introduce _any_ user-facing change?
    
    Yes. CrossValidatorModel size estimates no longer include shared runtime state retained by tuning configuration. Model behavior is unchanged.
    
    ### How was this patch tested?
    
    Added a CrossValidatorModel estimated-size regression test with a logger-initialized estimator.
    
    `JAVA_HOME=/usr/lib/jvm/java-17-openjdk-amd64 build/sbt 'mllib/testOnly org.apache.spark.ml.tuning.CrossValidatorSuite'`
    
    ### Was this patch authored or co-authored using generative AI tooling?
    
    Generated-by: Codex (GPT-5)
    
    Closes #57665 from zhengruifeng/tuning-model-size-metadata-dev3.
    
    Authored-by: Ruifeng Zheng <ru...@apache.org>
    Signed-off-by: Ruifeng Zheng <ru...@apache.org>
    (cherry picked from commit 8f5277839e3c0506e26aafbba6c9b9a26dc29c45)
    Signed-off-by: Ruifeng Zheng <ru...@apache.org>
---
 .../scala/org/apache/spark/ml/param/params.scala   | 31 ++++++++++++++++++++++
 .../apache/spark/ml/tuning/CrossValidator.scala    | 27 ++++++++++++++++++-
 .../spark/ml/tuning/CrossValidatorSuite.scala      | 16 +++++++++++
 3 files changed, 73 insertions(+), 1 deletion(-)

diff --git a/mllib/src/main/scala/org/apache/spark/ml/param/params.scala b/mllib/src/main/scala/org/apache/spark/ml/param/params.scala
index 21ccea479c1d..2e826e473e60 100644
--- a/mllib/src/main/scala/org/apache/spark/ml/param/params.scala
+++ b/mllib/src/main/scala/org/apache/spark/ml/param/params.scala
@@ -657,6 +657,37 @@ trait Params extends Identifiable with Serializable {
     SizeEstimator.estimate((this.paramMap, this.defaultParamMap, this.uid))
   }
 
+  /**
+   * Estimates metadata size while omitting params that can retain shared runtime state.
+   *
+   * A `Param` may hold this state indirectly through a validation closure that captures its
+   * owning instance, which can retain a `SparkSession`. Its value may also be a complex object,
+   * such as an `Estimator` in a `Param[Estimator[_]]`, that retains a `SparkSession`. Exclude
+   * such params to avoid accounting for shared infrastructure.
+   *
+   * @param excluded params to omit from the estimate
+   */
+  private[ml] def estimateMatadataSize(excluded: Seq[Param[_]]): Long = {
+    if (excluded.isEmpty) {
+      estimateMatadataSize
+    } else {
+      val filteredParamMap = mutable.Map.empty[Param[Any], Any]
+      paramMap.toSeq.foreach { pair =>
+        if (!excluded.contains(pair.param)) {
+          filteredParamMap(pair.param.asInstanceOf[Param[Any]]) = pair.value
+        }
+      }
+      val filteredDefaultParamMap = mutable.Map.empty[Param[Any], Any]
+      defaultParamMap.toSeq.foreach { pair =>
+        if (!excluded.contains(pair.param)) {
+          filteredDefaultParamMap(pair.param.asInstanceOf[Param[Any]]) = pair.value
+        }
+      }
+      SizeEstimator.estimate((
+        new ParamMap(filteredParamMap), new ParamMap(filteredDefaultParamMap), uid))
+    }
+  }
+
   /**
    * Returns all params sorted by their names. The default implementation uses Java reflection to
    * list all public methods that have no arguments and return [[Param]].
diff --git a/mllib/src/main/scala/org/apache/spark/ml/tuning/CrossValidator.scala b/mllib/src/main/scala/org/apache/spark/ml/tuning/CrossValidator.scala
index 119cc47a56f4..70bb2347cfc7 100644
--- a/mllib/src/main/scala/org/apache/spark/ml/tuning/CrossValidator.scala
+++ b/mllib/src/main/scala/org/apache/spark/ml/tuning/CrossValidator.scala
@@ -38,8 +38,8 @@ import org.apache.spark.ml.util.Instrumentation.instrumented
 import org.apache.spark.mllib.util.MLUtils
 import org.apache.spark.sql.{DataFrame, Dataset}
 import org.apache.spark.sql.types.{IntegerType, StructType}
+import org.apache.spark.util.{SizeEstimator, ThreadUtils}
 import org.apache.spark.util.ArrayImplicits._
-import org.apache.spark.util.ThreadUtils
 
 /**
  * Params for [[CrossValidator]] and [[CrossValidatorModel]].
@@ -328,6 +328,31 @@ class CrossValidatorModel private[ml] (
   @Since("2.3.0")
   def hasSubModels: Boolean = _subModels.isDefined
 
+  private[spark] override def estimatedSize: Long = {
+    var size = estimateMatadataSize(excluded = Seq(
+      // estimator: Param[Estimator[_]]
+      estimator,
+      // estimatorParamMaps: Param[Array[ParamMap]]
+      estimatorParamMaps,
+      // evaluator: Param[Evaluator]
+      evaluator))
+    // bestModel: Model[_]
+    size += bestModel.estimatedSize
+    // avgMetrics: Array[Double]
+    size += SizeEstimator.estimate(avgMetrics)
+    // _subModels: Option[Array[Array[Model[_]]]]
+    _subModels.foreach { models =>
+      models.foreach { modelArray =>
+        modelArray.foreach { model =>
+          if (model != null) {
+            size += model.estimatedSize
+          }
+        }
+      }
+    }
+    size
+  }
+
   @Since("2.0.0")
   override def transform(dataset: Dataset[_]): DataFrame = {
     transformSchema(dataset.schema, logging = true)
diff --git a/mllib/src/test/scala/org/apache/spark/ml/tuning/CrossValidatorSuite.scala b/mllib/src/test/scala/org/apache/spark/ml/tuning/CrossValidatorSuite.scala
index f97fefa24514..7673ec47c9f5 100644
--- a/mllib/src/test/scala/org/apache/spark/ml/tuning/CrossValidatorSuite.scala
+++ b/mllib/src/test/scala/org/apache/spark/ml/tuning/CrossValidatorSuite.scala
@@ -80,6 +80,22 @@ class CrossValidatorSuite
     }
   }
 
+  test("CrossValidatorModel estimated size") {
+    val estimator = new LogisticRegression().setMaxIter(1)
+    // Initialize the estimator's logger before retaining it in the CrossValidatorModel.
+    estimator.fit(dataset)
+    val model = new CrossValidator()
+      .setEstimator(estimator)
+      .setEstimatorParamMaps(Array(ParamMap.empty))
+      .setEvaluator(new BinaryClassificationEvaluator())
+      .setNumFolds(2)
+      .fit(dataset)
+
+    val maxSize = 16 * 1024
+    assert(model.estimatedSize < maxSize,
+      s"Estimation (${model.estimatedSize}) should not include shared runtime state")
+  }
+
   test("cross validation with logistic regression with fold col") {
     val lr = new LogisticRegression
     val lrParamMaps = new ParamGridBuilder()


---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@spark.apache.org
For additional commands, e-mail: commits-help@spark.apache.org