walterddr commented on a change in pull request #11420:
[FLINK-16565][python][ml] Make Pipeline Json compitable between Java and Python
if all PipelineStages are Java ones
URL: https://github.com/apache/flink/pull/11420#discussion_r393791750
##########
File path: flink-python/pyflink/ml/api/base.py
##########
@@ -265,11 +266,56 @@ def transform(self, t_env: TableEnvironment, input:
Table) -> Table:
return input
def to_json(self) -> str:
- import jsonpickle
- return str(jsonpickle.encode(self, keys=True))
+ """
+ If all PipelineStages in this Pipeline are Java ones, this method will
return a
+ Java json string, which can be loaded either from a Python Pipeline or
a Java Pipeline,
+ otherwise, it returns a Python json string which can only be loaded
from a Python Pipeline.
+ """
+ # if all PipelineStages are Java ones, we use Java toJson() to
generate Json string
+ # so that the string can also be loaded from Java side.
+ if all([type(stage) in [JavaTransformer, JavaEstimator, JavaModel]
+ for stage in self.get_stages()]):
+ j_pipeline =
get_gateway().jvm.org.apache.flink.ml.api.core.Pipeline()
+ for stage in self.get_stages():
+ stage._convert_params_to_java(stage._j_obj)
+ j_pipeline.appendStage(stage._j_obj)
+ return j_pipeline.toJson()
+ else:
+ import jsonpickle
+ return str(jsonpickle.encode(self, keys=True))
def load_json(self, json: str) -> None:
- import jsonpickle
- pipeline = jsonpickle.decode(json, keys=True)
- for stage in pipeline.get_stages():
- self.append_stage(stage)
+ """
+ This method can either load from a Java Pipeline json or a Python
Pipeline json.
+ """
+ # noinspection PyBroadException
+ try:
+ # try to load json with Python method
+ import jsonpickle
+ pipeline = jsonpickle.decode(json, keys=True)
+ for stage in pipeline.get_stages():
+ self.append_stage(stage)
+ except Exception:
+ # if can't load json with Python method, try to load with Java
method
+ gw = get_gateway()
+ j_pipeline = gw.jvm.org.apache.flink.ml.api.core.Pipeline()
+ j_pipeline.loadJson(json)
+
+ for j_stage in j_pipeline.getStages():
+ j_stage_class = j_stage.getClass()
+ j_transformer_class =
java_class(gw.jvm.org.apache.flink.ml.api.core.Transformer)
+ j_estimator_class =
java_class(gw.jvm.org.apache.flink.ml.api.core.Estimator)
+ j_model_class =
java_class(gw.jvm.org.apache.flink.ml.api.core.Model)
+ if j_transformer_class.isAssignableFrom(j_stage_class):
+ self.append_stage(JavaTransformer(j_stage))
+ elif j_estimator_class.isAssignableFrom(j_stage_class):
+ self.append_stage(JavaEstimator(j_stage))
+ elif j_model_class.isAssignableFrom(j_stage_class):
+ self.append_stage(JavaModel(j_stage))
+ else:
+ raise TypeError(
+ "Unexpected Java PipelineStage. It should be a %s, "
Review comment:
nit: makes error message more verbose, express which stage is problematic
```suggestion
"Unexpected Java PipelineStage: %s. Class should be
a %s, "
```
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
[email protected]
With regards,
Apache Git Services
