Revert "[SPARK-50922][ML][PYTHON][CONNECT] Support OneVsRest on Connect"

This reverts commit 22bac2e.

Author: HyukjinKwon

dev/sparktestsupport/modules.py

@@ -676,7 +676,6 @@ def __hash__(self):
         "pyspark.ml.tests.test_persistence",
         "pyspark.ml.tests.test_pipeline",
         "pyspark.ml.tests.test_tuning",
-        "pyspark.ml.tests.test_ovr",
         "pyspark.ml.tests.test_stat",
         "pyspark.ml.tests.test_training_summary",
         "pyspark.ml.tests.tuning.test_tuning",
@@ -1130,7 +1129,6 @@ def __hash__(self):
         "pyspark.ml.tests.connect.test_parity_feature",
         "pyspark.ml.tests.connect.test_parity_pipeline",
         "pyspark.ml.tests.connect.test_parity_tuning",
-        "pyspark.ml.tests.connect.test_parity_ovr",
     ],
     excluded_python_implementations=[
         "PyPy"  # Skip these tests under PyPy since they require numpy, pandas, and pyarrow and

python/pyspark/ml/classification.py

@@ -85,8 +85,6 @@
     MLWriter,
     MLWritable,
     HasTrainingSummary,
-    try_remote_read,
-    try_remote_write,
     try_remote_attribute_relation,
 )
 from pyspark.ml.wrapper import JavaParams, JavaPredictor, JavaPredictionModel, JavaWrapper
@@ -96,7 +94,6 @@
 from pyspark.sql.functions import udf, when
 from pyspark.sql.types import ArrayType, DoubleType
 from pyspark.storagelevel import StorageLevel
-from pyspark.sql.utils import is_remote
 
 if TYPE_CHECKING:
     from pyspark.ml._typing import P, ParamMap
@@ -3575,45 +3572,31 @@ def _fit(self, dataset: DataFrame) -> "OneVsRestModel":
         if handlePersistence:
             multiclassLabeled.persist(StorageLevel.MEMORY_AND_DISK)
 
-        def _oneClassFitTasks(numClasses: int):
-            indices = iter(range(numClasses))
-
-            def trainSingleClass() -> CM:
-                index = next(indices)
-
-                binaryLabelCol = "mc2b$" + str(index)
-                trainingDataset = multiclassLabeled.withColumn(
-                    binaryLabelCol,
-                    when(multiclassLabeled[labelCol] == float(index), 1.0).otherwise(0.0),
-                )
-                paramMap = dict(
-                    [
-                        (classifier.labelCol, binaryLabelCol),
-                        (classifier.featuresCol, featuresCol),
-                        (classifier.predictionCol, predictionCol),
-                    ]
-                )
-                if weightCol:
-                    paramMap[cast(HasWeightCol, classifier).weightCol] = weightCol
-                return index, classifier.fit(trainingDataset, paramMap)
-
-            return [trainSingleClass] * numClasses
+        def trainSingleClass(index: int) -> CM:
+            binaryLabelCol = "mc2b$" + str(index)
+            trainingDataset = multiclassLabeled.withColumn(
+                binaryLabelCol,
+                when(multiclassLabeled[labelCol] == float(index), 1.0).otherwise(0.0),
+            )
+            paramMap = dict(
+                [
+                    (classifier.labelCol, binaryLabelCol),
+                    (classifier.featuresCol, featuresCol),
+                    (classifier.predictionCol, predictionCol),
+                ]
+            )
+            if weightCol:
+                paramMap[cast(HasWeightCol, classifier).weightCol] = weightCol
+            return classifier.fit(trainingDataset, paramMap)
 
-        tasks = map(
-            inheritable_thread_target(dataset.sparkSession),
-            _oneClassFitTasks(numClasses),
-        )
         pool = ThreadPool(processes=min(self.getParallelism(), numClasses))
 
-        subModels = [None] * numClasses
-        for j, subModel in pool.imap_unordered(lambda f: f(), tasks):
-            assert subModels is not None
-            subModels[j] = subModel
+        models = pool.map(inheritable_thread_target(trainSingleClass), range(numClasses))
 
         if handlePersistence:
             multiclassLabeled.unpersist()
 
-        return self._copyValues(OneVsRestModel(models=subModels))
+        return self._copyValues(OneVsRestModel(models=models))
 
     def copy(self, extra: Optional["ParamMap"] = None) -> "OneVsRest":
         """
@@ -3688,11 +3671,9 @@ def _to_java(self) -> "JavaObject":
         return _java_obj
 
     @classmethod
-    @try_remote_read
     def read(cls) -> "OneVsRestReader":
         return OneVsRestReader(cls)
 
-    @try_remote_write
     def write(self) -> MLWriter:
         if isinstance(self.getClassifier(), JavaMLWritable):
             return JavaMLWriter(self)  # type: ignore[arg-type]
@@ -3806,7 +3787,7 @@ def __init__(self, models: List[ClassificationModel]):
         from pyspark.core.context import SparkContext
 
         self.models = models
-        if is_remote() or not isinstance(models[0], JavaMLWritable):
+        if not isinstance(models[0], JavaMLWritable):
             return
         # set java instance
         java_models = [cast(_JavaClassificationModel, model)._to_java() for model in self.models]
@@ -3974,11 +3955,9 @@ def _to_java(self) -> "JavaObject":
         return _java_obj
 
     @classmethod
-    @try_remote_read
     def read(cls) -> "OneVsRestModelReader":
         return OneVsRestModelReader(cls)
 
-    @try_remote_write
     def write(self) -> MLWriter:
         if all(
             map(

python/pyspark/ml/connect/readwrite.py

@@ -95,7 +95,6 @@ def saveInstance(
         from pyspark.ml.evaluation import JavaEvaluator
         from pyspark.ml.pipeline import Pipeline, PipelineModel
         from pyspark.ml.tuning import CrossValidator, TrainValidationSplit
-        from pyspark.ml.classification import OneVsRest, OneVsRestModel
 
         # Spark Connect ML is built on scala Spark.ML, that means we're only
         # supporting JavaModel or JavaEstimator or JavaEvaluator
@@ -188,27 +187,6 @@ def saveInstance(
                 warnings.warn("Overwrite doesn't take effect for TrainValidationSplitModel")
             tvsm_writer = RemoteTrainValidationSplitModelWriter(instance, optionMap, session)
             tvsm_writer.save(path)
-        elif isinstance(instance, OneVsRest):
-            from pyspark.ml.classification import OneVsRestWriter
-
-            if shouldOverwrite:
-                # TODO(SPARK-50954): Support client side model path overwrite
-                warnings.warn("Overwrite doesn't take effect for OneVsRest")
-
-            writer = OneVsRestWriter(instance)
-            writer.session(session)
-            writer.save(path)
-            # _OneVsRestSharedReadWrite.saveImpl(self.instance, self.sparkSession, path)
-        elif isinstance(instance, OneVsRestModel):
-            from pyspark.ml.classification import OneVsRestModelWriter
-
-            if shouldOverwrite:
-                # TODO(SPARK-50954): Support client side model path overwrite
-                warnings.warn("Overwrite doesn't take effect for OneVsRestModel")
-
-            writer = OneVsRestModelWriter(instance)
-            writer.session(session)
-            writer.save(path)
         else:
             raise NotImplementedError(f"Unsupported write for {instance.__class__}")
 
@@ -237,7 +215,6 @@ def loadInstance(
         from pyspark.ml.evaluation import JavaEvaluator
         from pyspark.ml.pipeline import Pipeline, PipelineModel
         from pyspark.ml.tuning import CrossValidator, TrainValidationSplit
-        from pyspark.ml.classification import OneVsRest, OneVsRestModel
 
         if (
             issubclass(clazz, JavaModel)
@@ -330,19 +307,5 @@ def _get_class() -> Type[RL]:
             tvs_reader.session(session)
             return tvs_reader.load(path)
 
-        elif issubclass(clazz, OneVsRest):
-            from pyspark.ml.classification import OneVsRestReader
-
-            ovr_reader = OneVsRestReader(OneVsRest)
-            ovr_reader.session(session)
-            return ovr_reader.load(path)
-
-        elif issubclass(clazz, OneVsRestModel):
-            from pyspark.ml.classification import OneVsRestModelReader
-
-            ovr_reader = OneVsRestModelReader(OneVsRestModel)
-            ovr_reader.session(session)
-            return ovr_reader.load(path)
-
         else:
             raise RuntimeError(f"Unsupported read for {clazz}")