You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@spark.apache.org by we...@apache.org on 2025/09/16 12:59:46 UTC
(spark) branch master updated: Fix: SparkML-connect can't load SparkML (legacy mode) saved model
This is an automated email from the ASF dual-hosted git repository.
weichenxu123 pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/spark.git
The following commit(s) were added to refs/heads/master by this push:
new f03c644ccfff Fix: SparkML-connect can't load SparkML (legacy mode) saved model
f03c644ccfff is described below
commit f03c644ccfff0b9836d446bb66f5c6425b29a473
Author: Weichen Xu <we...@databricks.com>
AuthorDate: Tue Sep 16 20:59:30 2025 +0800
Fix: SparkML-connect can't load SparkML (legacy mode) saved model
### What changes were proposed in this pull request?
Fix: SparkML-connect can't load SparkML (legacy mode) saved model
For meta algorithm like pipeline / cross-validator / one-vs-rest, if saving under **legacy Spark mode**, its writer goes through `JavaWriter` path but when loading under **Spark-connect mode**, its reader goes through `Python Reader` path,
`Python Reader` does not work with the data `JavaWriter` writes, because python reader requires metadata containing `language` field with `Python` value.
In this PR, I make the following changes:
* Make meta algorithm always use python reader, without checking metadata `language` field. Note that python reader is compatible with model written by JavaWriter, we don't need the metadata `language` field checking.
* Fix `_ValidatorParams` load methods: the `isJson` field and `value` field (non-json case) in `estimatorParamMaps` metadata should be deserialized from **JSON** string if it is saved by Java writer.
All model / estimator writer implementation code keeps intact.
### Why are the changes needed?
Bug fix.
### Does this PR introduce _any_ user-facing change?
No.
### How was this patch tested?
Manually.
### Was this patch authored or co-authored using generative AI tooling?
No.
Closes #52311 from WeichenXu123/SPARK-53555.
Authored-by: Weichen Xu <we...@databricks.com>
Signed-off-by: Weichen Xu <we...@databricks.com>
---
python/pyspark/ml/classification.py | 45 ++++----
python/pyspark/ml/pipeline.py | 16 +--
python/pyspark/ml/tuning.py | 202 +++++++++++++++++-------------------
3 files changed, 118 insertions(+), 145 deletions(-)
diff --git a/python/pyspark/ml/classification.py b/python/pyspark/ml/classification.py
index f66fc762971b..f5bfd59694af 100644
--- a/python/pyspark/ml/classification.py
+++ b/python/pyspark/ml/classification.py
@@ -78,7 +78,6 @@ from pyspark.ml.util import (
DefaultParamsReader,
DefaultParamsWriter,
JavaMLReadable,
- JavaMLReader,
JavaMLWritable,
JavaMLWriter,
MLReader,
@@ -3755,15 +3754,12 @@ class OneVsRestReader(MLReader[OneVsRest]):
def load(self, path: str) -> OneVsRest:
metadata = DefaultParamsReader.loadMetadata(path, self.sparkSession)
- if not DefaultParamsReader.isPythonParamsInstance(metadata):
- return JavaMLReader(self.cls).load(path) # type: ignore[arg-type]
- else:
- classifier = cast(
- Classifier, _OneVsRestSharedReadWrite.loadClassifier(path, self.sparkSession)
- )
- ova: OneVsRest = OneVsRest(classifier=classifier)._resetUid(metadata["uid"])
- DefaultParamsReader.getAndSetParams(ova, metadata, skipParams=["classifier"])
- return ova
+ classifier = cast(
+ Classifier, _OneVsRestSharedReadWrite.loadClassifier(path, self.sparkSession)
+ )
+ ova: OneVsRest = OneVsRest(classifier=classifier)._resetUid(metadata["uid"])
+ DefaultParamsReader.getAndSetParams(ova, metadata, skipParams=["classifier"])
+ return ova
@inherit_doc
@@ -3989,23 +3985,18 @@ class OneVsRestModelReader(MLReader[OneVsRestModel]):
def load(self, path: str) -> OneVsRestModel:
metadata = DefaultParamsReader.loadMetadata(path, self.sparkSession)
- if not DefaultParamsReader.isPythonParamsInstance(metadata):
- return JavaMLReader(self.cls).load(path) # type: ignore[arg-type]
- else:
- classifier = _OneVsRestSharedReadWrite.loadClassifier(path, self.sparkSession)
- numClasses = metadata["numClasses"]
- subModels = [None] * numClasses
- for idx in range(numClasses):
- subModelPath = os.path.join(path, f"model_{idx}")
- subModels[idx] = DefaultParamsReader.loadParamsInstance(
- subModelPath, self.sparkSession
- )
- ovaModel = OneVsRestModel(cast(List[ClassificationModel], subModels))._resetUid(
- metadata["uid"]
- )
- ovaModel.set(ovaModel.classifier, classifier)
- DefaultParamsReader.getAndSetParams(ovaModel, metadata, skipParams=["classifier"])
- return ovaModel
+ classifier = _OneVsRestSharedReadWrite.loadClassifier(path, self.sparkSession)
+ numClasses = metadata["numClasses"]
+ subModels = [None] * numClasses
+ for idx in range(numClasses):
+ subModelPath = os.path.join(path, f"model_{idx}")
+ subModels[idx] = DefaultParamsReader.loadParamsInstance(subModelPath, self.sparkSession)
+ ovaModel = OneVsRestModel(cast(List[ClassificationModel], subModels))._resetUid(
+ metadata["uid"]
+ )
+ ovaModel.set(ovaModel.classifier, classifier)
+ DefaultParamsReader.getAndSetParams(ovaModel, metadata, skipParams=["classifier"])
+ return ovaModel
@inherit_doc
diff --git a/python/pyspark/ml/pipeline.py b/python/pyspark/ml/pipeline.py
index b77392a50c7f..5728b8c0c511 100644
--- a/python/pyspark/ml/pipeline.py
+++ b/python/pyspark/ml/pipeline.py
@@ -25,12 +25,10 @@ from pyspark.ml.util import (
MLReadable,
MLWritable,
JavaMLWriter,
- JavaMLReader,
DefaultParamsReader,
DefaultParamsWriter,
MLWriter,
MLReader,
- JavaMLReadable,
JavaMLWritable,
try_remote_read,
try_remote_write,
@@ -250,11 +248,8 @@ class PipelineReader(MLReader[Pipeline]):
def load(self, path: str) -> Pipeline:
metadata = DefaultParamsReader.loadMetadata(path, self.sparkSession)
- if "language" not in metadata["paramMap"] or metadata["paramMap"]["language"] != "Python":
- return JavaMLReader(cast(Type["JavaMLReadable[Pipeline]"], self.cls)).load(path)
- else:
- uid, stages = PipelineSharedReadWrite.load(metadata, self.sparkSession, path)
- return Pipeline(stages=stages)._resetUid(uid)
+ uid, stages = PipelineSharedReadWrite.load(metadata, self.sparkSession, path)
+ return Pipeline(stages=stages)._resetUid(uid)
@inherit_doc
@@ -287,11 +282,8 @@ class PipelineModelReader(MLReader["PipelineModel"]):
def load(self, path: str) -> "PipelineModel":
metadata = DefaultParamsReader.loadMetadata(path, self.sparkSession)
- if "language" not in metadata["paramMap"] or metadata["paramMap"]["language"] != "Python":
- return JavaMLReader(cast(Type["JavaMLReadable[PipelineModel]"], self.cls)).load(path)
- else:
- uid, stages = PipelineSharedReadWrite.load(metadata, self.sparkSession, path)
- return PipelineModel(stages=cast(List[Transformer], stages))._resetUid(uid)
+ uid, stages = PipelineSharedReadWrite.load(metadata, self.sparkSession, path)
+ return PipelineModel(stages=cast(List[Transformer], stages))._resetUid(uid)
@inherit_doc
diff --git a/python/pyspark/ml/tuning.py b/python/pyspark/ml/tuning.py
index 498195baf1eb..d922ea38f2d8 100644
--- a/python/pyspark/ml/tuning.py
+++ b/python/pyspark/ml/tuning.py
@@ -15,6 +15,7 @@
# limitations under the License.
#
+import json
import os
import sys
import itertools
@@ -51,7 +52,6 @@ from pyspark.ml.util import (
MLReader,
MLWritable,
MLWriter,
- JavaMLReader,
JavaMLWriter,
try_remote_write,
try_remote_read,
@@ -415,7 +415,7 @@ class _ValidatorSharedReadWrite:
"is estimator, it cannot be meta estimator such as Validator or OneVsRest"
)
else:
- jsonParam["value"] = v
+ jsonParam["value"] = json.dumps(v)
jsonParam["isJson"] = True
jsonParamMap.append(jsonParam)
jsonEstimatorParamMaps.append(jsonParamMap)
@@ -444,16 +444,26 @@ class _ValidatorSharedReadWrite:
jsonEstimatorParamMaps = metadata["paramMap"]["estimatorParamMaps"]
+ is_saved_by_python_writer = DefaultParamsReader.isPythonParamsInstance(metadata)
+
estimatorParamMaps = []
for jsonParamMap in jsonEstimatorParamMaps:
paramMap = {}
for jsonParam in jsonParamMap:
est = uidToParams[jsonParam["parent"]]
param = getattr(est, jsonParam["name"])
- if "isJson" not in jsonParam or ("isJson" in jsonParam and jsonParam["isJson"]):
- value = jsonParam["value"]
+
+ def extract_value(key: str) -> Any:
+ if is_saved_by_python_writer:
+ return jsonParam[key]
+ # If the the params are serialized by java writer,
+ # the value is encoded as JSON string
+ return json.loads(jsonParam[key])
+
+ if "isJson" not in jsonParam or ("isJson" in jsonParam and extract_value("isJson")):
+ value = json.loads(jsonParam["value"])
else:
- relativePath = jsonParam["value"]
+ relativePath = extract_value("value")
valueSavedPath = os.path.join(path, relativePath)
value = DefaultParamsReader.loadParamsInstance(valueSavedPath, sc)
paramMap[param] = value
@@ -516,18 +526,15 @@ class CrossValidatorReader(MLReader["CrossValidator"]):
def load(self, path: str) -> "CrossValidator":
metadata = DefaultParamsReader.loadMetadata(path, self.sparkSession)
- if not DefaultParamsReader.isPythonParamsInstance(metadata):
- return JavaMLReader(self.cls).load(path) # type: ignore[arg-type]
- else:
- metadata, estimator, evaluator, estimatorParamMaps = _ValidatorSharedReadWrite.load(
- path, self.sparkSession, metadata
- )
- cv = CrossValidator(
- estimator=estimator, estimatorParamMaps=estimatorParamMaps, evaluator=evaluator
- )
- cv = cv._resetUid(metadata["uid"])
- DefaultParamsReader.getAndSetParams(cv, metadata, skipParams=["estimatorParamMaps"])
- return cv
+ metadata, estimator, evaluator, estimatorParamMaps = _ValidatorSharedReadWrite.load(
+ path, self.sparkSession, metadata
+ )
+ cv = CrossValidator(
+ estimator=estimator, estimatorParamMaps=estimatorParamMaps, evaluator=evaluator
+ )
+ cv = cv._resetUid(metadata["uid"])
+ DefaultParamsReader.getAndSetParams(cv, metadata, skipParams=["estimatorParamMaps"])
+ return cv
@inherit_doc
@@ -549,51 +556,44 @@ class CrossValidatorModelReader(MLReader["CrossValidatorModel"]):
def load(self, path: str) -> "CrossValidatorModel":
metadata = DefaultParamsReader.loadMetadata(path, self.sparkSession)
- if not DefaultParamsReader.isPythonParamsInstance(metadata):
- return JavaMLReader(self.cls).load(path) # type: ignore[arg-type]
+ metadata, estimator, evaluator, estimatorParamMaps = _ValidatorSharedReadWrite.load(
+ path, self.sparkSession, metadata
+ )
+ numFolds = metadata["paramMap"]["numFolds"]
+ bestModelPath = os.path.join(path, "bestModel")
+ bestModel: Model = DefaultParamsReader.loadParamsInstance(bestModelPath, self.sparkSession)
+ avgMetrics = metadata["avgMetrics"]
+ if "stdMetrics" in metadata:
+ stdMetrics = metadata["stdMetrics"]
else:
- metadata, estimator, evaluator, estimatorParamMaps = _ValidatorSharedReadWrite.load(
- path, self.sparkSession, metadata
- )
- numFolds = metadata["paramMap"]["numFolds"]
- bestModelPath = os.path.join(path, "bestModel")
- bestModel: Model = DefaultParamsReader.loadParamsInstance(
- bestModelPath, self.sparkSession
- )
- avgMetrics = metadata["avgMetrics"]
- if "stdMetrics" in metadata:
- stdMetrics = metadata["stdMetrics"]
- else:
- stdMetrics = None
- persistSubModels = ("persistSubModels" in metadata) and metadata["persistSubModels"]
-
- if persistSubModels:
- subModels = [[None] * len(estimatorParamMaps)] * numFolds
- for splitIndex in range(numFolds):
- for paramIndex in range(len(estimatorParamMaps)):
- modelPath = os.path.join(
- path, "subModels", f"fold{splitIndex}", f"{paramIndex}"
- )
- subModels[splitIndex][paramIndex] = DefaultParamsReader.loadParamsInstance(
- modelPath, self.sparkSession
- )
- else:
- subModels = None
+ stdMetrics = None
+ persistSubModels = ("persistSubModels" in metadata) and metadata["persistSubModels"]
- cvModel = CrossValidatorModel(
- bestModel,
- avgMetrics=avgMetrics,
- subModels=cast(List[List[Model]], subModels),
- stdMetrics=stdMetrics,
- )
- cvModel = cvModel._resetUid(metadata["uid"])
- cvModel.set(cvModel.estimator, estimator)
- cvModel.set(cvModel.estimatorParamMaps, estimatorParamMaps)
- cvModel.set(cvModel.evaluator, evaluator)
- DefaultParamsReader.getAndSetParams(
- cvModel, metadata, skipParams=["estimatorParamMaps"]
- )
- return cvModel
+ if persistSubModels:
+ subModels = [[None] * len(estimatorParamMaps)] * numFolds
+ for splitIndex in range(numFolds):
+ for paramIndex in range(len(estimatorParamMaps)):
+ modelPath = os.path.join(
+ path, "subModels", f"fold{splitIndex}", f"{paramIndex}"
+ )
+ subModels[splitIndex][paramIndex] = DefaultParamsReader.loadParamsInstance(
+ modelPath, self.sparkSession
+ )
+ else:
+ subModels = None
+
+ cvModel = CrossValidatorModel(
+ bestModel,
+ avgMetrics=avgMetrics,
+ subModels=cast(List[List[Model]], subModels),
+ stdMetrics=stdMetrics,
+ )
+ cvModel = cvModel._resetUid(metadata["uid"])
+ cvModel.set(cvModel.estimator, estimator)
+ cvModel.set(cvModel.estimatorParamMaps, estimatorParamMaps)
+ cvModel.set(cvModel.evaluator, evaluator)
+ DefaultParamsReader.getAndSetParams(cvModel, metadata, skipParams=["estimatorParamMaps"])
+ return cvModel
@inherit_doc
@@ -1194,18 +1194,15 @@ class TrainValidationSplitReader(MLReader["TrainValidationSplit"]):
def load(self, path: str) -> "TrainValidationSplit":
metadata = DefaultParamsReader.loadMetadata(path, self.sparkSession)
- if not DefaultParamsReader.isPythonParamsInstance(metadata):
- return JavaMLReader(self.cls).load(path) # type: ignore[arg-type]
- else:
- metadata, estimator, evaluator, estimatorParamMaps = _ValidatorSharedReadWrite.load(
- path, self.sparkSession, metadata
- )
- tvs = TrainValidationSplit(
- estimator=estimator, estimatorParamMaps=estimatorParamMaps, evaluator=evaluator
- )
- tvs = tvs._resetUid(metadata["uid"])
- DefaultParamsReader.getAndSetParams(tvs, metadata, skipParams=["estimatorParamMaps"])
- return tvs
+ metadata, estimator, evaluator, estimatorParamMaps = _ValidatorSharedReadWrite.load(
+ path, self.sparkSession, metadata
+ )
+ tvs = TrainValidationSplit(
+ estimator=estimator, estimatorParamMaps=estimatorParamMaps, evaluator=evaluator
+ )
+ tvs = tvs._resetUid(metadata["uid"])
+ DefaultParamsReader.getAndSetParams(tvs, metadata, skipParams=["estimatorParamMaps"])
+ return tvs
@inherit_doc
@@ -1227,42 +1224,35 @@ class TrainValidationSplitModelReader(MLReader["TrainValidationSplitModel"]):
def load(self, path: str) -> "TrainValidationSplitModel":
metadata = DefaultParamsReader.loadMetadata(path, self.sparkSession)
- if not DefaultParamsReader.isPythonParamsInstance(metadata):
- return JavaMLReader(self.cls).load(path) # type: ignore[arg-type]
- else:
- metadata, estimator, evaluator, estimatorParamMaps = _ValidatorSharedReadWrite.load(
- path, self.sparkSession, metadata
- )
- bestModelPath = os.path.join(path, "bestModel")
- bestModel: Model = DefaultParamsReader.loadParamsInstance(
- bestModelPath, self.sparkSession
- )
- validationMetrics = metadata["validationMetrics"]
- persistSubModels = ("persistSubModels" in metadata) and metadata["persistSubModels"]
+ metadata, estimator, evaluator, estimatorParamMaps = _ValidatorSharedReadWrite.load(
+ path, self.sparkSession, metadata
+ )
+ bestModelPath = os.path.join(path, "bestModel")
+ bestModel: Model = DefaultParamsReader.loadParamsInstance(bestModelPath, self.sparkSession)
+ validationMetrics = metadata["validationMetrics"]
+ persistSubModels = ("persistSubModels" in metadata) and metadata["persistSubModels"]
- if persistSubModels:
- subModels = [None] * len(estimatorParamMaps)
- for paramIndex in range(len(estimatorParamMaps)):
- modelPath = os.path.join(path, "subModels", f"{paramIndex}")
- subModels[paramIndex] = DefaultParamsReader.loadParamsInstance(
- modelPath, self.sparkSession
- )
- else:
- subModels = None
+ if persistSubModels:
+ subModels = [None] * len(estimatorParamMaps)
+ for paramIndex in range(len(estimatorParamMaps)):
+ modelPath = os.path.join(path, "subModels", f"{paramIndex}")
+ subModels[paramIndex] = DefaultParamsReader.loadParamsInstance(
+ modelPath, self.sparkSession
+ )
+ else:
+ subModels = None
- tvsModel = TrainValidationSplitModel(
- bestModel,
- validationMetrics=validationMetrics,
- subModels=cast(Optional[List[Model]], subModels),
- )
- tvsModel = tvsModel._resetUid(metadata["uid"])
- tvsModel.set(tvsModel.estimator, estimator)
- tvsModel.set(tvsModel.estimatorParamMaps, estimatorParamMaps)
- tvsModel.set(tvsModel.evaluator, evaluator)
- DefaultParamsReader.getAndSetParams(
- tvsModel, metadata, skipParams=["estimatorParamMaps"]
- )
- return tvsModel
+ tvsModel = TrainValidationSplitModel(
+ bestModel,
+ validationMetrics=validationMetrics,
+ subModels=cast(Optional[List[Model]], subModels),
+ )
+ tvsModel = tvsModel._resetUid(metadata["uid"])
+ tvsModel.set(tvsModel.estimator, estimator)
+ tvsModel.set(tvsModel.estimatorParamMaps, estimatorParamMaps)
+ tvsModel.set(tvsModel.evaluator, evaluator)
+ DefaultParamsReader.getAndSetParams(tvsModel, metadata, skipParams=["estimatorParamMaps"])
+ return tvsModel
@inherit_doc
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@spark.apache.org
For additional commands, e-mail: commits-help@spark.apache.org