You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@spark.apache.org by ru...@apache.org on 2025/02/02 01:23:43 UTC
(spark) branch branch-4.0 updated: [SPARK-50974][ML][PYTHON][CONNECT] Add support foldCol for CrossValidator on connect
This is an automated email from the ASF dual-hosted git repository.
ruifengz pushed a commit to branch branch-4.0
in repository https://gitbox.apache.org/repos/asf/spark.git
The following commit(s) were added to refs/heads/branch-4.0 by this push:
new 53844cdadd5f [SPARK-50974][ML][PYTHON][CONNECT] Add support foldCol for CrossValidator on connect
53844cdadd5f is described below
commit 53844cdadd5fc8237b42989e3bb62bc61dd3e793
Author: Bobby Wang <wb...@gmail.com>
AuthorDate: Sun Feb 2 09:23:16 2025 +0800
[SPARK-50974][ML][PYTHON][CONNECT] Add support foldCol for CrossValidator on connect
### What changes were proposed in this pull request?
This PR adds support foldCol for CrossValidator on connect
### Why are the changes needed?
feature parity
### Does this PR introduce _any_ user-facing change?
Yes
### How was this patch tested?
The CI passes
### Was this patch authored or co-authored using generative AI tooling?
No
Closes #49743 from wbo4958/cv-fold.
Authored-by: Bobby Wang <wb...@gmail.com>
Signed-off-by: Ruifeng Zheng <ru...@apache.org>
(cherry picked from commit 736752ec1ea0df69856b11b1452e21512443fc0e)
Signed-off-by: Ruifeng Zheng <ru...@apache.org>
---
python/pyspark/ml/tests/test_tuning.py | 107 ++++++++++++++++++++-
python/pyspark/ml/tests/tuning/test_tuning.py | 132 --------------------------
python/pyspark/ml/tuning.py | 22 ++++-
3 files changed, 119 insertions(+), 142 deletions(-)
diff --git a/python/pyspark/ml/tests/test_tuning.py b/python/pyspark/ml/tests/test_tuning.py
index 39b404d266fc..451c89db5b5a 100644
--- a/python/pyspark/ml/tests/test_tuning.py
+++ b/python/pyspark/ml/tests/test_tuning.py
@@ -31,6 +31,7 @@ from pyspark.ml.tuning import (
TrainValidationSplit,
TrainValidationSplitModel,
)
+from pyspark.sql.functions import rand
from pyspark.testing.sqlutils import ReusedSQLTestCase
@@ -64,7 +65,6 @@ class TuningTestsMixin:
# Access the train ratio
self.assertEqual(tvs_model.getTrainRatio(), 0.75)
- print("----------- ", tvs_model.validationMetrics)
self.assertTrue(np.isclose(tvs_model.validationMetrics[0], 0.5, atol=1e-4))
self.assertTrue(np.isclose(tvs_model.validationMetrics[1], 0.8857142857142857, atol=1e-4))
@@ -73,14 +73,14 @@ class TuningTestsMixin:
# save & load
with tempfile.TemporaryDirectory(prefix="train_validation_split") as d:
- path1 = os.path.join(d, "cv")
+ path1 = os.path.join(d, "tvs")
tvs.write().save(path1)
tvs2 = TrainValidationSplit.load(path1)
self.assertEqual(str(tvs), str(tvs2))
self.assertEqual(str(tvs.getEstimator()), str(tvs2.getEstimator()))
self.assertEqual(str(tvs.getEvaluator()), str(tvs2.getEvaluator()))
- path2 = os.path.join(d, "cv_model")
+ path2 = os.path.join(d, "tvsm")
tvs_model.write().save(path2)
model2 = TrainValidationSplitModel.load(path2)
self.assertEqual(str(tvs_model), str(model2))
@@ -103,7 +103,11 @@ class TuningTestsMixin:
grid = ParamGridBuilder().addGrid(lr.maxIter, [0, 1]).build()
evaluator = BinaryClassificationEvaluator()
cv = CrossValidator(
- estimator=lr, estimatorParamMaps=grid, evaluator=evaluator, parallelism=1
+ estimator=lr,
+ estimatorParamMaps=grid,
+ evaluator=evaluator,
+ parallelism=1,
+ collectSubModels=True,
)
self.assertEqual(cv.getEstimator(), lr)
@@ -123,8 +127,21 @@ class TuningTestsMixin:
)
self.assertEqual(output.count(), 50)
+ numFolds = cv.getNumFolds()
+
+ def checkSubModels(subModels):
+ self.assertEqual(len(subModels), numFolds)
+ for i in range(numFolds):
+ self.assertEqual(len(subModels[i]), len(grid))
+
+ checkSubModels(model.subModels)
+
+ for i in range(numFolds):
+ for j in range(len(grid)):
+ self.assertEqual(model.subModels[i][j].uid, model.subModels[i][j].uid)
+
# save & load
- with tempfile.TemporaryDirectory(prefix="cv_lr") as d:
+ with tempfile.TemporaryDirectory(prefix="cv") as d:
path1 = os.path.join(d, "cv")
cv.write().save(path1)
cv2 = CrossValidator.load(path1)
@@ -135,10 +152,90 @@ class TuningTestsMixin:
path2 = os.path.join(d, "cv_model")
model.write().save(path2)
model2 = CrossValidatorModel.load(path2)
+ checkSubModels(model2.subModels)
self.assertEqual(str(model), str(model2))
self.assertEqual(str(model.getEstimator()), str(model2.getEstimator()))
self.assertEqual(str(model.getEvaluator()), str(model2.getEvaluator()))
+ path2 = os.path.join(d, "cv_model2")
+ model.write().option("persistSubModels", "false").save(path2)
+ cvModel2 = CrossValidatorModel.load(path2)
+ self.assertEqual(cvModel2.subModels, None)
+
+ model3 = model2.copy()
+ checkSubModels(model3.subModels)
+
+ def test_cv_user_specified_folds(self):
+ from pyspark.sql import functions as F
+
+ dataset = self.spark.createDataFrame(
+ [
+ (Vectors.dense([0.0]), 0.0),
+ (Vectors.dense([0.4]), 1.0),
+ (Vectors.dense([0.5]), 0.0),
+ (Vectors.dense([0.6]), 1.0),
+ (Vectors.dense([1.0]), 1.0),
+ ]
+ * 10,
+ ["features", "label"],
+ ).repartition(2, "features")
+
+ dataset_with_folds = (
+ dataset.repartition(1)
+ .withColumn("random", rand(100))
+ .withColumn(
+ "fold",
+ F.when(F.col("random") < 0.33, 0).when(F.col("random") < 0.66, 1).otherwise(2),
+ )
+ .repartition(2, "features")
+ )
+
+ lr = LogisticRegression()
+ grid = ParamGridBuilder().addGrid(lr.maxIter, [20]).build()
+ evaluator = BinaryClassificationEvaluator()
+
+ cv = CrossValidator(estimator=lr, estimatorParamMaps=grid, evaluator=evaluator, numFolds=3)
+ cv_with_user_folds = CrossValidator(
+ estimator=lr, estimatorParamMaps=grid, evaluator=evaluator, numFolds=3, foldCol="fold"
+ )
+
+ self.assertEqual(cv.getEstimator().uid, cv_with_user_folds.getEstimator().uid)
+
+ cvModel1 = cv.fit(dataset)
+ cvModel2 = cv_with_user_folds.fit(dataset_with_folds)
+ for index in range(len(cvModel1.avgMetrics)):
+ print(abs(cvModel1.avgMetrics[index] - cvModel2.avgMetrics[index]))
+ self.assertTrue(abs(cvModel1.avgMetrics[index] - cvModel2.avgMetrics[index]) < 0.1)
+
+ def test_cv_invalid_user_specified_folds(self):
+ dataset_with_folds = self.spark.createDataFrame(
+ [
+ (Vectors.dense([0.0]), 0.0, 0),
+ (Vectors.dense([0.4]), 1.0, 1),
+ (Vectors.dense([0.5]), 0.0, 2),
+ (Vectors.dense([0.6]), 1.0, 0),
+ (Vectors.dense([1.0]), 1.0, 1),
+ ]
+ * 10,
+ ["features", "label", "fold"],
+ )
+
+ lr = LogisticRegression()
+ grid = ParamGridBuilder().addGrid(lr.maxIter, [20]).build()
+ evaluator = BinaryClassificationEvaluator()
+
+ cv = CrossValidator(
+ estimator=lr, estimatorParamMaps=grid, evaluator=evaluator, numFolds=2, foldCol="fold"
+ )
+ with self.assertRaisesRegex(Exception, "Fold number must be in range"):
+ cv.fit(dataset_with_folds)
+
+ cv = CrossValidator(
+ estimator=lr, estimatorParamMaps=grid, evaluator=evaluator, numFolds=4, foldCol="fold"
+ )
+ with self.assertRaisesRegex(Exception, "The validation data at fold 3 is empty"):
+ cv.fit(dataset_with_folds)
+
class TuningTests(TuningTestsMixin, ReusedSQLTestCase):
pass
diff --git a/python/pyspark/ml/tests/tuning/test_tuning.py b/python/pyspark/ml/tests/tuning/test_tuning.py
index 83acedc4b581..cbbe7b6cca82 100644
--- a/python/pyspark/ml/tests/tuning/test_tuning.py
+++ b/python/pyspark/ml/tests/tuning/test_tuning.py
@@ -30,7 +30,6 @@ from pyspark.ml.linalg import Vectors
from pyspark.ml.param import Param, Params
from pyspark.ml.tuning import (
CrossValidator,
- CrossValidatorModel,
ParamGridBuilder,
TrainValidationSplit,
TrainValidationSplitModel,
@@ -233,137 +232,6 @@ class CrossValidatorTests(SparkSessionTestCase, ValidatorTestUtilsMixin):
self.assertEqual(cvSerialModel.avgMetrics, cvParallelModel.avgMetrics)
self.assertEqual(cvSerialModel.stdMetrics, cvParallelModel.stdMetrics)
- def test_expose_sub_models(self):
- temp_path = tempfile.mkdtemp()
- dataset = self.spark.createDataFrame(
- [
- (Vectors.dense([0.0]), 0.0),
- (Vectors.dense([0.4]), 1.0),
- (Vectors.dense([0.5]), 0.0),
- (Vectors.dense([0.6]), 1.0),
- (Vectors.dense([1.0]), 1.0),
- ]
- * 10,
- ["features", "label"],
- )
-
- lr = LogisticRegression()
- grid = ParamGridBuilder().addGrid(lr.maxIter, [0, 1]).build()
- evaluator = BinaryClassificationEvaluator()
-
- numFolds = 3
- cv = CrossValidator(
- estimator=lr,
- estimatorParamMaps=grid,
- evaluator=evaluator,
- numFolds=numFolds,
- collectSubModels=True,
- )
-
- def checkSubModels(subModels):
- self.assertEqual(len(subModels), numFolds)
- for i in range(numFolds):
- self.assertEqual(len(subModels[i]), len(grid))
-
- cvModel = cv.fit(dataset)
- checkSubModels(cvModel.subModels)
-
- # Test the default value for option "persistSubModel" to be "true"
- testSubPath = temp_path + "/testCrossValidatorSubModels"
- savingPathWithSubModels = testSubPath + "cvModel3"
- cvModel.save(savingPathWithSubModels)
- cvModel3 = CrossValidatorModel.load(savingPathWithSubModels)
- checkSubModels(cvModel3.subModels)
- cvModel4 = cvModel3.copy()
- checkSubModels(cvModel4.subModels)
-
- savingPathWithoutSubModels = testSubPath + "cvModel2"
- cvModel.write().option("persistSubModels", "false").save(savingPathWithoutSubModels)
- cvModel2 = CrossValidatorModel.load(savingPathWithoutSubModels)
- self.assertEqual(cvModel2.subModels, None)
-
- for i in range(numFolds):
- for j in range(len(grid)):
- self.assertEqual(cvModel.subModels[i][j].uid, cvModel3.subModels[i][j].uid)
-
- def test_user_specified_folds(self):
- from pyspark.sql import functions as F
-
- dataset = self.spark.createDataFrame(
- [
- (Vectors.dense([0.0]), 0.0),
- (Vectors.dense([0.4]), 1.0),
- (Vectors.dense([0.5]), 0.0),
- (Vectors.dense([0.6]), 1.0),
- (Vectors.dense([1.0]), 1.0),
- ]
- * 10,
- ["features", "label"],
- ).repartition(2, "features")
-
- dataset_with_folds = (
- dataset.repartition(1)
- .withColumn("random", rand(100))
- .withColumn(
- "fold",
- F.when(F.col("random") < 0.33, 0).when(F.col("random") < 0.66, 1).otherwise(2),
- )
- .repartition(2, "features")
- )
-
- lr = LogisticRegression()
- grid = ParamGridBuilder().addGrid(lr.maxIter, [20]).build()
- evaluator = BinaryClassificationEvaluator()
-
- cv = CrossValidator(estimator=lr, estimatorParamMaps=grid, evaluator=evaluator, numFolds=3)
- cv_with_user_folds = CrossValidator(
- estimator=lr, estimatorParamMaps=grid, evaluator=evaluator, numFolds=3, foldCol="fold"
- )
-
- self.assertEqual(cv.getEstimator().uid, cv_with_user_folds.getEstimator().uid)
-
- cvModel1 = cv.fit(dataset)
- cvModel2 = cv_with_user_folds.fit(dataset_with_folds)
- for index in range(len(cvModel1.avgMetrics)):
- print(abs(cvModel1.avgMetrics[index] - cvModel2.avgMetrics[index]))
- self.assertTrue(abs(cvModel1.avgMetrics[index] - cvModel2.avgMetrics[index]) < 0.1)
-
- # test save/load of CrossValidator
- temp_path = tempfile.mkdtemp()
- cvPath = temp_path + "/cv"
- cv_with_user_folds.save(cvPath)
- loadedCV = CrossValidator.load(cvPath)
- self.assertEqual(loadedCV.getFoldCol(), cv_with_user_folds.getFoldCol())
-
- def test_invalid_user_specified_folds(self):
- dataset_with_folds = self.spark.createDataFrame(
- [
- (Vectors.dense([0.0]), 0.0, 0),
- (Vectors.dense([0.4]), 1.0, 1),
- (Vectors.dense([0.5]), 0.0, 2),
- (Vectors.dense([0.6]), 1.0, 0),
- (Vectors.dense([1.0]), 1.0, 1),
- ]
- * 10,
- ["features", "label", "fold"],
- )
-
- lr = LogisticRegression()
- grid = ParamGridBuilder().addGrid(lr.maxIter, [20]).build()
- evaluator = BinaryClassificationEvaluator()
-
- cv = CrossValidator(
- estimator=lr, estimatorParamMaps=grid, evaluator=evaluator, numFolds=2, foldCol="fold"
- )
- with self.assertRaisesRegex(Exception, "Fold number must be in range"):
- cv.fit(dataset_with_folds)
-
- cv = CrossValidator(
- estimator=lr, estimatorParamMaps=grid, evaluator=evaluator, numFolds=4, foldCol="fold"
- )
- with self.assertRaisesRegex(Exception, "The validation data at fold 3 is empty"):
- cv.fit(dataset_with_folds)
-
class TrainValidationSplitTests(SparkSessionTestCase, ValidatorTestUtilsMixin):
def test_fit_minimize_metric(self):
diff --git a/python/pyspark/ml/tuning.py b/python/pyspark/ml/tuning.py
index e27ef955ef73..d3506bf1c6b0 100644
--- a/python/pyspark/ml/tuning.py
+++ b/python/pyspark/ml/tuning.py
@@ -55,9 +55,10 @@ from pyspark.ml.util import (
JavaMLWriter,
try_remote_write,
try_remote_read,
+ is_remote,
)
from pyspark.ml.wrapper import JavaParams, JavaEstimator, JavaWrapper
-from pyspark.sql.functions import col, lit, rand, UserDefinedFunction
+from pyspark.sql.functions import col, lit, rand
from pyspark.sql.types import BooleanType
from pyspark.sql.dataframe import DataFrame
@@ -901,16 +902,27 @@ class CrossValidator(
)
return True
+ if is_remote():
+ from pyspark.sql.connect.udf import UserDefinedFunction
+ else:
+ from pyspark.sql.functions import UserDefinedFunction # type: ignore[assignment]
+
checker_udf = UserDefinedFunction(checker, BooleanType())
for i in range(nFolds):
training = dataset.filter(checker_udf(dataset[foldCol]) & (col(foldCol) != lit(i)))
validation = dataset.filter(
checker_udf(dataset[foldCol]) & (col(foldCol) == lit(i))
)
- if training.rdd.getNumPartitions() == 0 or len(training.take(1)) == 0:
- raise ValueError("The training data at fold %s is empty." % i)
- if validation.rdd.getNumPartitions() == 0 or len(validation.take(1)) == 0:
- raise ValueError("The validation data at fold %s is empty." % i)
+ if is_remote():
+ if len(training.take(1)) == 0:
+ raise ValueError("The training data at fold %s is empty." % i)
+ if len(validation.take(1)) == 0:
+ raise ValueError("The validation data at fold %s is empty." % i)
+ else:
+ if training.rdd.getNumPartitions() == 0 or len(training.take(1)) == 0:
+ raise ValueError("The training data at fold %s is empty." % i)
+ if validation.rdd.getNumPartitions() == 0 or len(validation.take(1)) == 0:
+ raise ValueError("The validation data at fold %s is empty." % i)
datasets.append((training, validation))
return datasets
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@spark.apache.org
For additional commands, e-mail: commits-help@spark.apache.org