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 2026/08/12 08:28:27 UTC
(spark) branch branch-4.x updated: [SPARK-58733][PYTHON][ML][TEST] Revert test_cv_io_pipeline.py parallelization
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 98b85168d8b5 [SPARK-58733][PYTHON][ML][TEST] Revert test_cv_io_pipeline.py parallelization
98b85168d8b5 is described below
commit 98b85168d8b52021666c16f18a712ac03e75ce87
Author: Ruifeng Zheng <ru...@apache.org>
AuthorDate: Wed Aug 12 16:27:59 2026 +0800
[SPARK-58733][PYTHON][ML][TEST] Revert test_cv_io_pipeline.py parallelization
### What changes were proposed in this pull request?
This reverts the test parallelization introduced by #53725. It runs the real and dummy
logistic-regression variants sequentially and restores the default parallelism for
`CrossValidator` and `OneVsRest`.
### Why are the changes needed?
The test combines an outer two-worker thread pool with `CrossValidator(parallelism=4)` and
`OneVsRest(parallelism=2)`. This is flaky in resource-sensitive cases: the nested concurrency can
exhaust the available JVM heap and fail with `java.lang.OutOfMemoryError: Java heap space` while
Spark is serializing a broadcast. The persistence coverage does not require concurrent model
training, so sequential execution is more reliable.
### Does this PR introduce _any_ user-facing change?
No.
### How was this patch tested?
Syntax and formatting checks:
```
git diff --check
conda run -n spark-dev-313 python -m py_compile \
python/pyspark/ml/tests/tuning/test_cv_io_pipeline.py
```
The focused PySpark test was not run locally because this checkout did not have assembled Spark
artifacts. It will be covered by the PySpark ML GitHub Actions job.
### Was this patch authored or co-authored using generative AI tooling?
Generated-by: OpenAI Codex (GPT-5)
Closes #57835 from zhengruifeng/SPARK-54958-revert-cv-pipeline-parallelism.
Authored-by: Ruifeng Zheng <ru...@apache.org>
Signed-off-by: Ruifeng Zheng <ru...@foxmail.com>
(cherry picked from commit 3429eeb9115c0355dad29d0daeea7b12dd4a6874)
Signed-off-by: Ruifeng Zheng <ru...@foxmail.com>
---
python/pyspark/ml/tests/tuning/test_cv_io_pipeline.py | 14 +++-----------
1 file changed, 3 insertions(+), 11 deletions(-)
diff --git a/python/pyspark/ml/tests/tuning/test_cv_io_pipeline.py b/python/pyspark/ml/tests/tuning/test_cv_io_pipeline.py
index a4eefc98f6b4..32e97ceb4f84 100644
--- a/python/pyspark/ml/tests/tuning/test_cv_io_pipeline.py
+++ b/python/pyspark/ml/tests/tuning/test_cv_io_pipeline.py
@@ -16,7 +16,6 @@
#
import tempfile
-from concurrent.futures import ThreadPoolExecutor
from pyspark.ml.feature import HashingTF, Tokenizer
from pyspark.ml import Pipeline
@@ -55,7 +54,7 @@ class CrossValidatorIOPipelineTests(SparkSessionTestCase, ValidatorTestUtilsMixi
tokenizer = Tokenizer(inputCol="text", outputCol="words")
hashingTF = HashingTF(inputCol=tokenizer.getOutputCol(), outputCol="features")
- ova = OneVsRest(classifier=LogisticRegressionCls(), parallelism=2)
+ ova = OneVsRest(classifier=LogisticRegressionCls())
lr1 = LogisticRegressionCls().setMaxIter(5)
lr2 = LogisticRegressionCls().setMaxIter(10)
@@ -73,7 +72,6 @@ class CrossValidatorIOPipelineTests(SparkSessionTestCase, ValidatorTestUtilsMixi
estimatorParamMaps=paramGrid,
evaluator=MulticlassClassificationEvaluator(),
numFolds=2,
- parallelism=4,
) # use 3+ folds in practice
cvPath = temp_path + "/cv"
crossval.save(cvPath)
@@ -102,7 +100,6 @@ class CrossValidatorIOPipelineTests(SparkSessionTestCase, ValidatorTestUtilsMixi
estimatorParamMaps=paramGrid,
evaluator=MulticlassClassificationEvaluator(),
numFolds=2,
- parallelism=4,
) # use 3+ folds in practice
cv2Path = temp_path + "/cv2"
crossval2.save(cv2Path)
@@ -129,13 +126,8 @@ class CrossValidatorIOPipelineTests(SparkSessionTestCase, ValidatorTestUtilsMixi
self.assertEqual(loadedStage.uid, originalStage.uid)
def test_save_load_pipeline_estimator(self):
- with ThreadPoolExecutor(max_workers=2) as executor:
- list(
- executor.map(
- self._run_test_save_load_pipeline_estimator,
- [LogisticRegression, DummyLogisticRegression],
- )
- )
+ self._run_test_save_load_pipeline_estimator(LogisticRegression)
+ self._run_test_save_load_pipeline_estimator(DummyLogisticRegression)
if __name__ == "__main__":
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@spark.apache.org
For additional commands, e-mail: commits-help@spark.apache.org