You are viewing a plain text version of this content. The canonical link for it is here.
Posted to reviews@spark.apache.org by "WeichenXu123 (via GitHub)" <gi...@apache.org> on 2023/02/13 02:11:19 UTC
[GitHub] [spark] WeichenXu123 opened a new pull request, #39985: [WIP] Initial prototype implementation of PySpark ML via Spark connect
WeichenXu123 opened a new pull request, #39985:
URL: https://github.com/apache/spark/pull/39985
<!--
Thanks for sending a pull request! Here are some tips for you:
1. If this is your first time, please read our contributor guidelines: https://spark.apache.org/contributing.html
2. Ensure you have added or run the appropriate tests for your PR: https://spark.apache.org/developer-tools.html
3. If the PR is unfinished, add '[WIP]' in your PR title, e.g., '[WIP][SPARK-XXXX] Your PR title ...'.
4. Be sure to keep the PR description updated to reflect all changes.
5. Please write your PR title to summarize what this PR proposes.
6. If possible, provide a concise example to reproduce the issue for a faster review.
7. If you want to add a new configuration, please read the guideline first for naming configurations in
'core/src/main/scala/org/apache/spark/internal/config/ConfigEntry.scala'.
8. If you want to add or modify an error type or message, please read the guideline first in
'core/src/main/resources/error/README.md'.
-->
### What changes were proposed in this pull request?
<!--
Please clarify what changes you are proposing. The purpose of this section is to outline the changes and how this PR fixes the issue.
If possible, please consider writing useful notes for better and faster reviews in your PR. See the examples below.
1. If you refactor some codes with changing classes, showing the class hierarchy will help reviewers.
2. If you fix some SQL features, you can provide some references of other DBMSes.
3. If there is design documentation, please add the link.
4. If there is a discussion in the mailing list, please add the link.
-->
### Why are the changes needed?
<!--
Please clarify why the changes are needed. For instance,
1. If you propose a new API, clarify the use case for a new API.
2. If you fix a bug, you can clarify why it is a bug.
-->
### Does this PR introduce _any_ user-facing change?
<!--
Note that it means *any* user-facing change including all aspects such as the documentation fix.
If yes, please clarify the previous behavior and the change this PR proposes - provide the console output, description and/or an example to show the behavior difference if possible.
If possible, please also clarify if this is a user-facing change compared to the released Spark versions or within the unreleased branches such as master.
If no, write 'No'.
-->
### How was this patch tested?
<!--
If tests were added, say they were added here. Please make sure to add some test cases that check the changes thoroughly including negative and positive cases if possible.
If it was tested in a way different from regular unit tests, please clarify how you tested step by step, ideally copy and paste-able, so that other reviewers can test and check, and descendants can verify in the future.
If tests were not added, please describe why they were not added and/or why it was difficult to add.
If benchmark tests were added, please run the benchmarks in GitHub Actions for the consistent environment, and the instructions could accord to: https://spark.apache.org/developer-tools.html#github-workflow-benchmarks.
-->
--
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.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] WeichenXu123 commented on a diff in pull request #39985: [SPARK-42412][WIP] Initial prototype implementation of PySpark ML via Spark connect
Posted by "WeichenXu123 (via GitHub)" <gi...@apache.org>.
WeichenXu123 commented on code in PR #39985:
URL: https://github.com/apache/spark/pull/39985#discussion_r1107183606
##########
connector/connect/common/src/main/protobuf/spark/connect/ml.proto:
##########
@@ -0,0 +1,166 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+syntax = 'proto3';
+
+package spark.connect;
+
+import "spark/connect/relations.proto";
+
+option java_multiple_files = true;
+option java_package = "org.apache.spark.connect.proto";
+
+
+message RemoteCall {
+ oneof call_type {
+ ConstructObject construct_object = 1;
+ DestructObject destruct_object = 2;
+ CallMethod call_method = 3;
+ CallFunction call_function = 4;
+ }
Review Comment:
@grundprinzip
PR is updated.
The first stage of this PR is complete, it can support all java-backend estimator's fit / transform / all model attributes / model summary / model saving loading.
I haven't tested pyspark Pipeline, but I guess it is already supported directly without other change.
You can test it by the code pasted in PR description.
TODO list see my PR description.
I will attach design doc later.
The final goal is to seamlessly support all pyspark ML estimators / transformers without any user code changes and keep the same behavior with non-spark-connect mode. With current design, we are not far from the goal and we don't need to change too much code.
--
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.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] WeichenXu123 commented on a diff in pull request #39985: [SPARK-42412][WIP] Initial prototype implementation of PySpark ML via Spark connect
Posted by "WeichenXu123 (via GitHub)" <gi...@apache.org>.
WeichenXu123 commented on code in PR #39985:
URL: https://github.com/apache/spark/pull/39985#discussion_r1106973688
##########
connector/connect/common/src/main/protobuf/spark/connect/ml.proto:
##########
@@ -0,0 +1,166 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+syntax = 'proto3';
+
+package spark.connect;
+
+import "spark/connect/relations.proto";
+
+option java_multiple_files = true;
+option java_package = "org.apache.spark.connect.proto";
+
+
+message RemoteCall {
+ oneof call_type {
+ ConstructObject construct_object = 1;
+ DestructObject destruct_object = 2;
+ CallMethod call_method = 3;
+ CallFunction call_function = 4;
+ }
+
+ message ArgValue {
Review Comment:
```
int32 int32_value = 1;
int64 int64_value = 2;
float float_value = 3;
double double_value = 4;
bool bool_value = 5;
string string_value = 6;
```
for these fields above, I agree we can turn them into a `Expression.Literal` message.
--
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.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] WeichenXu123 commented on a diff in pull request #39985: [SPARK-42412][WIP] Initial prototype implementation of PySpark ML via Spark connect
Posted by "WeichenXu123 (via GitHub)" <gi...@apache.org>.
WeichenXu123 commented on code in PR #39985:
URL: https://github.com/apache/spark/pull/39985#discussion_r1106973688
##########
connector/connect/common/src/main/protobuf/spark/connect/ml.proto:
##########
@@ -0,0 +1,166 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+syntax = 'proto3';
+
+package spark.connect;
+
+import "spark/connect/relations.proto";
+
+option java_multiple_files = true;
+option java_package = "org.apache.spark.connect.proto";
+
+
+message RemoteCall {
+ oneof call_type {
+ ConstructObject construct_object = 1;
+ DestructObject destruct_object = 2;
+ CallMethod call_method = 3;
+ CallFunction call_function = 4;
+ }
+
+ message ArgValue {
Review Comment:
```
int32 int32_value = 1;
int64 int64_value = 2;
float float_value = 3;
double double_value = 4;
bool bool_value = 5;
string string_value = 6;
```
for these fields above, I agree we can turn them into a `Literal` message.
--
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.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] WeichenXu123 commented on a diff in pull request #39985: [SPARK-42412][WIP] Initial prototype implementation of PySpark ML via Spark connect
Posted by "WeichenXu123 (via GitHub)" <gi...@apache.org>.
WeichenXu123 commented on code in PR #39985:
URL: https://github.com/apache/spark/pull/39985#discussion_r1106968910
##########
connector/connect/server/src/main/scala/org/apache/spark/sql/connect/planner/MLUtils.scala:
##########
@@ -0,0 +1,79 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.spark.sql.connect.planner
+
+import org.apache.spark.ml.param.{Param, Params}
+
+object SparkConnectUtils {
+
+ private def castParamValue(paramType: Class[_], paramValue: Object): Object = {
+ if (paramType == classOf[Int]) {
+ java.lang.Integer.valueOf(paramValue.asInstanceOf[java.lang.Number].intValue())
+ } else if (paramType == classOf[Long]) {
+ java.lang.Long.valueOf(paramValue.asInstanceOf[java.lang.Number].longValue())
+ } else if (paramType == classOf[Float]) {
+ java.lang.Float.valueOf(paramValue.asInstanceOf[java.lang.Number].floatValue())
+ } else if (paramType == classOf[Double]) {
+ java.lang.Double.valueOf(paramValue.asInstanceOf[java.lang.Number].doubleValue())
+ } else if (paramType == classOf[Array[Int]]) {
+ paramValue.asInstanceOf[List[Object]]
+ .map(castParamValue(classOf[Int], _).asInstanceOf[java.lang.Integer])
+ .map(_.intValue()).toArray
+ } else if (paramType == classOf[Array[Long]]) {
+ paramValue.asInstanceOf[List[Object]]
+ .map(castParamValue(classOf[Long], _).asInstanceOf[java.lang.Long])
+ .map(_.longValue()).toArray
+ } else if (paramType == classOf[Array[Float]]) {
+ paramValue.asInstanceOf[List[Object]]
+ .map(castParamValue(classOf[Float], _).asInstanceOf[java.lang.Float])
+ .map(_.floatValue()).toArray
+ } else if (paramType == classOf[Array[Double]]) {
+ paramValue.asInstanceOf[List[Object]]
+ .map(castParamValue(classOf[Double], _).asInstanceOf[java.lang.Double])
+ .map(_.doubleValue()).toArray
+ } else if (paramType == classOf[Array[Boolean]]) {
+ paramValue.asInstanceOf[List[java.lang.Boolean]].map(_.booleanValue()).toArray
+ } else if (paramType == classOf[Array[String]]) {
+ paramValue.asInstanceOf[List[String]].toArray
+ } else {
+ throw new RuntimeException()
Review Comment:
there's a bug here actually, I fixed it.
--
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.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] WeichenXu123 commented on a diff in pull request #39985: [SPARK-42412][WIP] Initial prototype implementation of PySpark ML via Spark connect
Posted by "WeichenXu123 (via GitHub)" <gi...@apache.org>.
WeichenXu123 commented on code in PR #39985:
URL: https://github.com/apache/spark/pull/39985#discussion_r1107893116
##########
connector/connect/common/src/main/protobuf/spark/connect/ml.proto:
##########
@@ -0,0 +1,166 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+syntax = 'proto3';
+
+package spark.connect;
+
+import "spark/connect/relations.proto";
+
+option java_multiple_files = true;
+option java_package = "org.apache.spark.connect.proto";
+
+
+message RemoteCall {
+ oneof call_type {
+ ConstructObject construct_object = 1;
+ DestructObject destruct_object = 2;
+ CallMethod call_method = 3;
+ CallFunction call_function = 4;
+ }
Review Comment:
sure! I will share design doc later for current design intention explanation and list other interface options for discussion, and this prototying PR is requested by @mengxr for proving the project feasible.
--
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.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] srowen commented on a diff in pull request #39985: [SPARK-42412][WIP] Initial prototype implementation of PySpark ML via Spark connect
Posted by "srowen (via GitHub)" <gi...@apache.org>.
srowen commented on code in PR #39985:
URL: https://github.com/apache/spark/pull/39985#discussion_r1106181670
##########
connector/connect/server/src/main/scala/org/apache/spark/sql/connect/planner/MLUtils.scala:
##########
@@ -0,0 +1,79 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.spark.sql.connect.planner
+
+import org.apache.spark.ml.param.{Param, Params}
+
+object SparkConnectUtils {
+
+ private def castParamValue(paramType: Class[_], paramValue: Object): Object = {
+ if (paramType == classOf[Int]) {
Review Comment:
Probably the answer is no, but can't this be done with `match`?
##########
connector/connect/server/src/main/scala/org/apache/spark/sql/connect/planner/MLUtils.scala:
##########
@@ -0,0 +1,79 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.spark.sql.connect.planner
+
+import org.apache.spark.ml.param.{Param, Params}
+
+object SparkConnectUtils {
+
+ private def castParamValue(paramType: Class[_], paramValue: Object): Object = {
+ if (paramType == classOf[Int]) {
+ java.lang.Integer.valueOf(paramValue.asInstanceOf[java.lang.Number].intValue())
+ } else if (paramType == classOf[Long]) {
+ java.lang.Long.valueOf(paramValue.asInstanceOf[java.lang.Number].longValue())
+ } else if (paramType == classOf[Float]) {
+ java.lang.Float.valueOf(paramValue.asInstanceOf[java.lang.Number].floatValue())
+ } else if (paramType == classOf[Double]) {
+ java.lang.Double.valueOf(paramValue.asInstanceOf[java.lang.Number].doubleValue())
Review Comment:
How about Byte? we probably need Array[Byte] at least
##########
connector/connect/server/src/main/scala/org/apache/spark/sql/connect/planner/MLUtils.scala:
##########
@@ -0,0 +1,79 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.spark.sql.connect.planner
+
+import org.apache.spark.ml.param.{Param, Params}
+
+object SparkConnectUtils {
+
+ private def castParamValue(paramType: Class[_], paramValue: Object): Object = {
+ if (paramType == classOf[Int]) {
+ java.lang.Integer.valueOf(paramValue.asInstanceOf[java.lang.Number].intValue())
+ } else if (paramType == classOf[Long]) {
+ java.lang.Long.valueOf(paramValue.asInstanceOf[java.lang.Number].longValue())
+ } else if (paramType == classOf[Float]) {
+ java.lang.Float.valueOf(paramValue.asInstanceOf[java.lang.Number].floatValue())
+ } else if (paramType == classOf[Double]) {
+ java.lang.Double.valueOf(paramValue.asInstanceOf[java.lang.Number].doubleValue())
+ } else if (paramType == classOf[Array[Int]]) {
+ paramValue.asInstanceOf[List[Object]]
+ .map(castParamValue(classOf[Int], _).asInstanceOf[java.lang.Integer])
Review Comment:
Nit, why not one map() statement doing both?
##########
connector/connect/server/src/main/scala/org/apache/spark/sql/connect/planner/MLUtils.scala:
##########
@@ -0,0 +1,79 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.spark.sql.connect.planner
+
+import org.apache.spark.ml.param.{Param, Params}
+
+object SparkConnectUtils {
+
+ private def castParamValue(paramType: Class[_], paramValue: Object): Object = {
+ if (paramType == classOf[Int]) {
+ java.lang.Integer.valueOf(paramValue.asInstanceOf[java.lang.Number].intValue())
+ } else if (paramType == classOf[Long]) {
+ java.lang.Long.valueOf(paramValue.asInstanceOf[java.lang.Number].longValue())
+ } else if (paramType == classOf[Float]) {
+ java.lang.Float.valueOf(paramValue.asInstanceOf[java.lang.Number].floatValue())
+ } else if (paramType == classOf[Double]) {
+ java.lang.Double.valueOf(paramValue.asInstanceOf[java.lang.Number].doubleValue())
+ } else if (paramType == classOf[Array[Int]]) {
+ paramValue.asInstanceOf[List[Object]]
+ .map(castParamValue(classOf[Int], _).asInstanceOf[java.lang.Integer])
+ .map(_.intValue()).toArray
+ } else if (paramType == classOf[Array[Long]]) {
+ paramValue.asInstanceOf[List[Object]]
+ .map(castParamValue(classOf[Long], _).asInstanceOf[java.lang.Long])
+ .map(_.longValue()).toArray
+ } else if (paramType == classOf[Array[Float]]) {
+ paramValue.asInstanceOf[List[Object]]
+ .map(castParamValue(classOf[Float], _).asInstanceOf[java.lang.Float])
+ .map(_.floatValue()).toArray
+ } else if (paramType == classOf[Array[Double]]) {
+ paramValue.asInstanceOf[List[Object]]
+ .map(castParamValue(classOf[Double], _).asInstanceOf[java.lang.Double])
+ .map(_.doubleValue()).toArray
+ } else if (paramType == classOf[Array[Boolean]]) {
+ paramValue.asInstanceOf[List[java.lang.Boolean]].map(_.booleanValue()).toArray
+ } else if (paramType == classOf[Array[String]]) {
+ paramValue.asInstanceOf[List[String]].toArray
+ } else {
+ throw new RuntimeException()
Review Comment:
At least here say the requested type, and the value
##########
connector/connect/server/src/main/scala/org/apache/spark/sql/connect/planner/SparkMLPlanner.scala:
##########
@@ -0,0 +1,402 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.sql.connect.planner
+
+import io.grpc.stub.StreamObserver
+import java.lang
+import scala.reflect.runtime.universe
+
+import org.apache.spark.connect.proto
+import org.apache.spark.connect.proto.{ExecutePlanRequest, ExecutePlanResponse, RemoteCall}
+import org.apache.spark.sql.{DataFrame, Dataset}
+import org.apache.spark.sql.connect.service.{SessionHolder}
+import org.apache.spark.util.Utils
+
+object SparkMLPlanner {
+
+ val runtimeMirror = universe.runtimeMirror(Utils.getContextOrSparkClassLoader)
+
+ def buildRemoteCallResponse(
+ request: ExecutePlanRequest,
+ responseObserver: StreamObserver[ExecutePlanResponse],
+ returnValueOpt: Option[RemoteCall.ArgValue]
+ ): Unit = {
+ val response = proto.ExecutePlanResponse.newBuilder
+ .setClientId(request.getClientId)
+
+ returnValueOpt.foreach(response.setRemoteCallReturnValue(_))
Review Comment:
Total nit, you can omit (_) usually
##########
connector/connect/server/src/main/scala/org/apache/spark/sql/connect/planner/MLUtils.scala:
##########
@@ -0,0 +1,79 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.spark.sql.connect.planner
+
+import org.apache.spark.ml.param.{Param, Params}
+
+object SparkConnectUtils {
+
+ private def castParamValue(paramType: Class[_], paramValue: Object): Object = {
+ if (paramType == classOf[Int]) {
+ java.lang.Integer.valueOf(paramValue.asInstanceOf[java.lang.Number].intValue())
+ } else if (paramType == classOf[Long]) {
+ java.lang.Long.valueOf(paramValue.asInstanceOf[java.lang.Number].longValue())
+ } else if (paramType == classOf[Float]) {
+ java.lang.Float.valueOf(paramValue.asInstanceOf[java.lang.Number].floatValue())
+ } else if (paramType == classOf[Double]) {
+ java.lang.Double.valueOf(paramValue.asInstanceOf[java.lang.Number].doubleValue())
+ } else if (paramType == classOf[Array[Int]]) {
+ paramValue.asInstanceOf[List[Object]]
+ .map(castParamValue(classOf[Int], _).asInstanceOf[java.lang.Integer])
+ .map(_.intValue()).toArray
+ } else if (paramType == classOf[Array[Long]]) {
+ paramValue.asInstanceOf[List[Object]]
+ .map(castParamValue(classOf[Long], _).asInstanceOf[java.lang.Long])
+ .map(_.longValue()).toArray
+ } else if (paramType == classOf[Array[Float]]) {
+ paramValue.asInstanceOf[List[Object]]
+ .map(castParamValue(classOf[Float], _).asInstanceOf[java.lang.Float])
+ .map(_.floatValue()).toArray
+ } else if (paramType == classOf[Array[Double]]) {
+ paramValue.asInstanceOf[List[Object]]
+ .map(castParamValue(classOf[Double], _).asInstanceOf[java.lang.Double])
+ .map(_.doubleValue()).toArray
+ } else if (paramType == classOf[Array[Boolean]]) {
+ paramValue.asInstanceOf[List[java.lang.Boolean]].map(_.booleanValue()).toArray
+ } else if (paramType == classOf[Array[String]]) {
+ paramValue.asInstanceOf[List[String]].toArray
+ } else {
+ throw new RuntimeException()
+ }
+ }
+
+ private def makeParamPair(
+ instance: Params, paramName: String, paramValue: Object
Review Comment:
Nit, wrong wrapping
##########
connector/connect/server/src/main/scala/org/apache/spark/sql/connect/planner/SparkMLPlanner.scala:
##########
@@ -0,0 +1,402 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.sql.connect.planner
+
+import io.grpc.stub.StreamObserver
+import java.lang
+import scala.reflect.runtime.universe
+
+import org.apache.spark.connect.proto
+import org.apache.spark.connect.proto.{ExecutePlanRequest, ExecutePlanResponse, RemoteCall}
+import org.apache.spark.sql.{DataFrame, Dataset}
+import org.apache.spark.sql.connect.service.{SessionHolder}
+import org.apache.spark.util.Utils
+
+object SparkMLPlanner {
+
+ val runtimeMirror = universe.runtimeMirror(Utils.getContextOrSparkClassLoader)
+
+ def buildRemoteCallResponse(
+ request: ExecutePlanRequest,
+ responseObserver: StreamObserver[ExecutePlanResponse],
+ returnValueOpt: Option[RemoteCall.ArgValue]
+ ): Unit = {
+ val response = proto.ExecutePlanResponse.newBuilder
+ .setClientId(request.getClientId)
+
+ returnValueOpt.foreach(response.setRemoteCallReturnValue(_))
+
+ responseObserver.onNext(response.build())
+ responseObserver.onCompleted()
+ }
+
+ def loadInputRelation(sessionHolder: SessionHolder, inputRelation: proto.Relation): DataFrame = {
+ val relationalPlanner = new SparkConnectPlanner(sessionHolder)
+ val plan = relationalPlanner.transformRelation(inputRelation)
+ Dataset.ofRows(sessionHolder.session, plan)
+ }
+
+ /*
+ def createStage(uid: String, className: String): PipelineStage = {
+ val clazz = Utils.classForName(className)
+ val ctorOpt = clazz.getConstructors.find { ctor =>
+ // Find the constructor with signature `this(uid: String)`
+ ctor.getParameterCount == 1 && ctor.getParameterTypes()(0).eq(classOf[String])
+ }
+ if (ctorOpt.isEmpty) {
+ throw new RuntimeException(
+ s"Could not find 'this(uid: String)' constructor for class $className"
+ )
+ }
+
+ ctorOpt.get.newInstance(uid).asInstanceOf[PipelineStage]
+ }
+
+ def extractParamPair(
+ instance: Params,
+ paramName: String,
+ protoValue: MlCommand.ParamValue
+ ): (Param[Any], Any) = {
+ import MlCommand.ParamValue.ValueCase
+
+ val param = instance.getParam(paramName)
+ val valueType = param.paramValueClassTag.runtimeClass
+ val extractedValue = protoValue.getValueCase match {
+ case ValueCase.INT_VAL =>
+ assert(valueType == classOf[Int])
+ protoValue.getIntVal
+
+ case ValueCase.LONG_VAL =>
+ if (valueType == classOf[Long]) {
+ protoValue.getLongVal
+ } else if (valueType == classOf[Int]) {
+ protoValue.getLongVal.toInt
+ } else {
+ throw new java.lang.AssertionError()
+ }
+
+ case ValueCase.FLOAT_VAL =>
+ assert(valueType == classOf[Float])
+ protoValue.getFloatVal
+
+ case ValueCase.DOUBLE_VAL =>
+ if (valueType == classOf[Double]) {
+ protoValue.getDoubleVal
+ } else if (valueType == classOf[Float]) {
+ protoValue.getDoubleVal.toFloat
+ } else {
+ throw new java.lang.AssertionError()
+ }
+
+ case ValueCase.BOOL_VAL =>
+ assert(valueType == classOf[Boolean])
+ protoValue.getBoolVal
+
+ case ValueCase.STR_VAL =>
+ assert(valueType == classOf[String])
+ protoValue.getStrVal
+
+ case ValueCase.INT_ARRAY =>
+ assert(valueType == classOf[Array[Int]])
+ protoValue.getIntArray.getElementList.stream().mapToInt(_.toInt).toArray
+
+ case ValueCase.LONG_ARRAY =>
+ val elemList = protoValue.getLongArray.getElementList
+ if (valueType == classOf[Array[Long]]) {
+ elemList.stream().mapToLong(_.toLong).toArray
+ } else if (valueType == classOf[Array[Int]]) {
+ elemList.stream().mapToInt(_.toInt).toArray
+ } else {
+ throw new java.lang.AssertionError()
+ }
+
+ case ValueCase.FLOAT_ARRAY =>
+ assert(valueType == classOf[Float])
+ val floatList = protoValue.getFloatArray.getElementList
+ val floatArray = new Array[Float](floatList.size())
+ for (i <- 0 until floatList.size()) {
+ floatArray(i) = floatList.get(i)
+ }
+ floatArray
+
+ case ValueCase.DOUBLE_ARRAY =>
+ val doubleList = protoValue.getDoubleArray.getElementList
+ if (valueType == classOf[Array[Double]]) {
+ doubleList.stream().mapToDouble(_.toDouble).toArray
+ } else if (valueType == classOf[Array[Float]]) {
+ val floatArray = new Array[Float](doubleList.size())
+ for (i <- 0 until doubleList.size()) {
+ floatArray(i) = doubleList.get(i).toFloat
+ }
+ floatArray
+ } else {
+ throw new java.lang.AssertionError()
+ }
+
+ case ValueCase.STR_ARRAY =>
+ assert(valueType == classOf[Array[String]])
+ protoValue.getStrArray.getElementList.toArray(Array[String]())
+
+ case ValueCase.BOOL_ARRAY =>
+ assert(valueType == classOf[Array[Boolean]])
+ val boolList = protoValue.getBoolArray.getElementList
+ val boolArray = new Array[Boolean](boolList.size())
+ for (i <- 0 until boolList.size()) {
+ boolArray(i) = boolList.get(i)
+ }
+ boolArray
+
+ case _ =>
+ throw InvalidPlanInput()
+ }
+ (param, extractedValue)
+ }
+
+ def setParams(instance: Params, protoParams: MlCommand.Params): Unit = {
+ protoParams.getParamsMap.forEach { (paramName, protoValue) =>
+ val (param, paramValue) = extractParamPair(
+ instance, paramName, protoValue
+ )
+ instance.set(param, paramValue)
+ }
+ protoParams.getDefaultParamsMap.forEach { (paramName, protoValue) =>
+ val (param, paramValue) = extractParamPair(
+ instance, paramName, protoValue
+ )
+ instance._setDefault(param, paramValue)
+ }
+ }
+
+ def handleRemoteCall(
+ sessionHolder: SessionHolder,
+ request: ExecutePlanRequest,
+ responseObserver: StreamObserver[ExecutePlanResponse]
+ ): Unit = {
+ val mlCommand = request.getPlan.getMlCommand
+ mlCommand.getOpTypeCase match {
+ case proto.MlCommand.OpTypeCase.CONSTRUCT_STAGE =>
+ val constructStage = mlCommand.getConstructStage
+ val stage = createStage(
+ constructStage.getUid,
+ constructStage.getClassName
+ )
+ val objectId = serverSideObjectManager.registerObject(stage)
+
+ val resp = ExecutePlanResponse.MlCommandResponse
+ .newBuilder.setServerSideObjectId(objectId).build()
+
+ buildMlResponse(request, responseObserver, Some(resp))
+
+ case proto.MlCommand.OpTypeCase.DESTRUCT_OBJECT =>
+ serverSideObjectManager.removeObject(mlCommand.getDestructObject.getId)
+
+ case proto.MlCommand.OpTypeCase.FIT =>
+ val estimator = serverSideObjectManager
+ .getObject(mlCommand.getFit.getId).asInstanceOf[Estimator[_]]
+ val inputDF = loadInputRelation(session, mlCommand.getFit.getInput)
+ val model = estimator.fit(inputDF)
+ val modelObjId = serverSideObjectManager.
+ registerObject(model.asInstanceOf[Object])
+
+ val resp = ExecutePlanResponse.MlCommandResponse
+ .newBuilder.setServerSideObjectId(modelObjId).build()
+
+ buildMlResponse(request, responseObserver, Some(resp))
+
+ case proto.MlCommand.OpTypeCase.TRANSFORM =>
+ val transformer = serverSideObjectManager.getObject(mlCommand.getTransform.getId)
+ .asInstanceOf[Transformer]
+ val inputDF = loadInputRelation(session, mlCommand.getTransform.getInput)
+ val transformedDF = transformer.transform(inputDF)
+ val transformedDFId = serverSideObjectManager.registerObject(transformedDF)
+
+ val resp = ExecutePlanResponse.MlCommandResponse
+ .newBuilder.setServerSideObjectId(transformedDFId).build()
+
+ buildMlResponse(request, responseObserver, Some(resp))
+
+ case proto.MlCommand.OpTypeCase.TRANSFER_PARAMS_TO_SERVER =>
+ val instance = serverSideObjectManager.getObject(mlCommand.getTransferParamsToServer.getId)
+ .asInstanceOf[Params]
+ val protoParams = mlCommand.getTransferParamsToServer.getParams
+ setParams(instance, protoParams)
+
+ buildMlResponse(request, responseObserver, None)
+
+ case _ =>
+ throw new UnsupportedOperationException(s"${mlCommand.getOpTypeCase} not supported.")
+ }
+ }
+ */
+
+ def parseArg(protoArg: proto.RemoteCall.ArgValue, sessionHolder: SessionHolder): Object = {
+ import scala.collection.JavaConverters._
+ import proto.RemoteCall.ArgValue
+ protoArg.getArgTypeCase match {
+ case ArgValue.ArgTypeCase.INT32_VALUE =>
+ lang.Integer.valueOf(protoArg.getInt32Value)
+ case ArgValue.ArgTypeCase.INT64_VALUE =>
+ lang.Long.valueOf(protoArg.getInt64Value)
+ case ArgValue.ArgTypeCase.FLOAT_VALUE =>
+ lang.Float.valueOf(protoArg.getFloatValue)
+ case ArgValue.ArgTypeCase.DOUBLE_VALUE =>
+ lang.Double.valueOf(protoArg.getDoubleValue)
+ case ArgValue.ArgTypeCase.STRING_VALUE =>
+ protoArg.getStringValue
+ case ArgValue.ArgTypeCase.BOOL_VALUE =>
+ lang.Boolean.valueOf(protoArg.getBoolValue)
+ case ArgValue.ArgTypeCase.RELATION =>
+ loadInputRelation(sessionHolder, protoArg.getRelation)
+ case ArgValue.ArgTypeCase.REMOTE_OBJECT =>
+ sessionHolder.serverSideObjectManager.getObject(protoArg.getRemoteObject.getId)
+ case ArgValue.ArgTypeCase.LIST =>
+ val protoValues = protoArg.getList.getElementList.asScala.toList
+ protoValues.map { x => parseArg(x, sessionHolder) }
+ case ArgValue.ArgTypeCase.MAP =>
+ val protoMap = protoArg.getMap.getMapMap().asScala.toMap
+ protoMap.mapValues(x => parseArg(x, sessionHolder))
+ case _ =>
+ throw InvalidPlanInput()
+ }
+ }
+
+ /**
+ * Parse prototype message of argument list, returns a tuple of (values, types)
+ */
+ def parseArgs(
+ protoArgList: List[proto.RemoteCall.ArgValue],
+ sessionHolder: SessionHolder
+ ): Array[Object] = {
+ val argValues = new Array[Object](protoArgList.size)
+
+ for (i <- 0 until protoArgList.size) {
+ val protoValue = protoArgList(i)
+ argValues(i) = parseArg(protoValue, sessionHolder)
+ }
+ argValues
+ }
+
+ def serializeValue(value: Object, sessionHolder: SessionHolder): proto.RemoteCall.ArgValue = {
+ val protoBuilder = proto.RemoteCall.ArgValue.newBuilder()
+ value match {
+ case v: lang.Integer =>
+ protoBuilder.setInt32Value(v.intValue())
+ case v: lang.Long =>
+ protoBuilder.setInt64Value(v.longValue())
+ case v: lang.Float =>
+ protoBuilder.setFloatValue(v.floatValue())
+ case v: lang.Double =>
+ protoBuilder.setDoubleValue(v.doubleValue())
+ case v: String =>
+ protoBuilder.setStringValue(v)
+ case v: lang.Boolean =>
+ protoBuilder.setBoolValue(v)
+ case v: List[_] =>
+ // TODO
+ throw new UnsupportedOperationException()
+ case v: Map[_, _] =>
+ // TODO
+ throw new UnsupportedOperationException()
+ case v: Object =>
+ val instanceId = sessionHolder.serverSideObjectManager.registerObject(v)
+ protoBuilder.setRemoteObject(
+ RemoteCall.RemoteObject.newBuilder()
+ .setId(instanceId)
+ .build()
+ )
+ }
+ protoBuilder.build()
+ }
+
+ def invokeMethod(
+ instance: Object,
+ methodName: String,
+ argsProto: List[RemoteCall.ArgValue],
+ sessionHolder: SessionHolder): Option[RemoteCall.ArgValue] = {
+ val argValues = parseArgs(argsProto, sessionHolder)
+ val method = instance.getClass.getMethods().find { method =>
Review Comment:
I wonder if this will eventually be a bottleneck for method invocation - maybe not. We could cache results, but that's complicating things
--
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.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] WeichenXu123 commented on a diff in pull request #39985: [SPARK-42412][WIP] Initial prototype implementation of PySpark ML via Spark connect
Posted by "WeichenXu123 (via GitHub)" <gi...@apache.org>.
WeichenXu123 commented on code in PR #39985:
URL: https://github.com/apache/spark/pull/39985#discussion_r1106969912
##########
connector/connect/server/src/main/scala/org/apache/spark/sql/connect/planner/MLUtils.scala:
##########
@@ -0,0 +1,79 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.spark.sql.connect.planner
+
+import org.apache.spark.ml.param.{Param, Params}
+
+object SparkConnectUtils {
+
+ private def castParamValue(paramType: Class[_], paramValue: Object): Object = {
+ if (paramType == classOf[Int]) {
+ java.lang.Integer.valueOf(paramValue.asInstanceOf[java.lang.Number].intValue())
+ } else if (paramType == classOf[Long]) {
+ java.lang.Long.valueOf(paramValue.asInstanceOf[java.lang.Number].longValue())
+ } else if (paramType == classOf[Float]) {
+ java.lang.Float.valueOf(paramValue.asInstanceOf[java.lang.Number].floatValue())
+ } else if (paramType == classOf[Double]) {
+ java.lang.Double.valueOf(paramValue.asInstanceOf[java.lang.Number].doubleValue())
Review Comment:
Yes I will add support for it.
--
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.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] amaliujia commented on a diff in pull request #39985: [SPARK-42412][WIP] Initial prototype implementation of PySpark ML via Spark connect
Posted by "amaliujia (via GitHub)" <gi...@apache.org>.
amaliujia commented on code in PR #39985:
URL: https://github.com/apache/spark/pull/39985#discussion_r1107685447
##########
connector/connect/common/src/main/protobuf/spark/connect/ml.proto:
##########
@@ -0,0 +1,166 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+syntax = 'proto3';
+
+package spark.connect;
+
+import "spark/connect/relations.proto";
+
+option java_multiple_files = true;
+option java_package = "org.apache.spark.connect.proto";
+
+
+message RemoteCall {
+ oneof call_type {
+ ConstructObject construct_object = 1;
+ DestructObject destruct_object = 2;
+ CallMethod call_method = 3;
+ CallFunction call_function = 4;
+ }
+
+ message ArgValue {
Review Comment:
For something like
```
RemoteObject remote_object = 10;
Vector vector = 11;
Matrix matrix = 12;
```
Are you suggesting to extend the `Expression.Literal` to include these or just replace those compatible types with `Expression.Literal` but keep things like `Vector` and `Matrix` in ml.proto?
--
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.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] WeichenXu123 commented on a diff in pull request #39985: [SPARK-42412][WIP] Initial prototype implementation of PySpark ML via Spark connect
Posted by "WeichenXu123 (via GitHub)" <gi...@apache.org>.
WeichenXu123 commented on code in PR #39985:
URL: https://github.com/apache/spark/pull/39985#discussion_r1106564992
##########
connector/connect/common/src/main/protobuf/spark/connect/ml.proto:
##########
@@ -0,0 +1,166 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+syntax = 'proto3';
+
+package spark.connect;
+
+import "spark/connect/relations.proto";
+
+option java_multiple_files = true;
+option java_package = "org.apache.spark.connect.proto";
+
+
+message RemoteCall {
+ oneof call_type {
+ ConstructObject construct_object = 1;
+ DestructObject destruct_object = 2;
+ CallMethod call_method = 3;
+ CallFunction call_function = 4;
+ }
Review Comment:
Sorry for the confusion! The PR is not complete, I will complete it today and attach a runnable ML case demo and attach design sketch.
--
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.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] WeichenXu123 commented on a diff in pull request #39985: [SPARK-42412][WIP] Initial prototype implementation of PySpark ML via Spark connect
Posted by "WeichenXu123 (via GitHub)" <gi...@apache.org>.
WeichenXu123 commented on code in PR #39985:
URL: https://github.com/apache/spark/pull/39985#discussion_r1107183606
##########
connector/connect/common/src/main/protobuf/spark/connect/ml.proto:
##########
@@ -0,0 +1,166 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+syntax = 'proto3';
+
+package spark.connect;
+
+import "spark/connect/relations.proto";
+
+option java_multiple_files = true;
+option java_package = "org.apache.spark.connect.proto";
+
+
+message RemoteCall {
+ oneof call_type {
+ ConstructObject construct_object = 1;
+ DestructObject destruct_object = 2;
+ CallMethod call_method = 3;
+ CallFunction call_function = 4;
+ }
Review Comment:
@grundprinzip
PR is updated.
The first stage of this PR is complete, it can support all java-backend estimator's fit / transform / all model attributes / model summary.
I haven't tested pyspark Pipeline, but I guess it is already supported directly without other change.
You can test it by the code pasted in PR description.
TODO:
* Add whitelist checking for remote call.
* Support pyspark CrossValidator and other meta-estimator
* Support pyspark evaluator
* remote object GC
* Some minor fixes (e.g. set model params / model uid correctly)
* add test.
I will attach design doc later.
--
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.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] grundprinzip commented on a diff in pull request #39985: [SPARK-42412][WIP] Initial prototype implementation of PySpark ML via Spark connect
Posted by "grundprinzip (via GitHub)" <gi...@apache.org>.
grundprinzip commented on code in PR #39985:
URL: https://github.com/apache/spark/pull/39985#discussion_r1107692251
##########
connector/connect/common/src/main/protobuf/spark/connect/ml.proto:
##########
@@ -0,0 +1,166 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+syntax = 'proto3';
+
+package spark.connect;
+
+import "spark/connect/relations.proto";
+
+option java_multiple_files = true;
+option java_package = "org.apache.spark.connect.proto";
+
+
+message RemoteCall {
+ oneof call_type {
+ ConstructObject construct_object = 1;
+ DestructObject destruct_object = 2;
+ CallMethod call_method = 3;
+ CallFunction call_function = 4;
+ }
Review Comment:
As said before the current proto spec does to explain the intent of the message it's purely an interface to construct jvm classes on the other side. The goal for Spark Connect is not to have a 1:1 mapping of JVM method calls to Spark Connect RPC invocations but operate on a higher level abstraction.
This becomes apparent when you're looking at the code that you're modifying on the Python side, you're explicitly mimicking the Py4J call interface.
Before you invest a lot of time into building a Py4J over GRPC lets have a better discussion on what the interface should look like, please!
Thanks for the hard work!
--
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.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] github-actions[bot] closed pull request #39985: [SPARK-42412][WIP] Initial prototype implementation of PySpark ML via Spark connect
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] closed pull request #39985: [SPARK-42412][WIP] Initial prototype implementation of PySpark ML via Spark connect
URL: https://github.com/apache/spark/pull/39985
--
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.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] WeichenXu123 commented on a diff in pull request #39985: [SPARK-42412][WIP] Initial prototype implementation of PySpark ML via Spark connect
Posted by "WeichenXu123 (via GitHub)" <gi...@apache.org>.
WeichenXu123 commented on code in PR #39985:
URL: https://github.com/apache/spark/pull/39985#discussion_r1104451588
##########
connector/connect/common/src/main/protobuf/spark/connect/ml.proto:
##########
@@ -0,0 +1,165 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+syntax = 'proto3';
+
+package spark.connect;
+
+import "spark/connect/relations.proto";
+
+option java_multiple_files = true;
+option java_package = "org.apache.spark.connect.proto";
+
+
+message RemoteCall {
+ oneof call_type {
+ ConstructObject construct_object = 1;
+ DestructObject destruct_object = 2;
+ CallMethod call_method = 3;
+ CallFunction call_function = 4;
+ }
+
+ message ArgValue {
+ oneof arg_type {
+ int32 int32_value = 1;
+ int64 int64_value = 2;
+ float float_value = 3;
+ double double_value = 4;
+ bool bool_value = 5;
+ string string_value = 6;
+ List list = 7;
+ Map map = 8;
+ Relation relation = 9;
+ RemoteObject remote_object = 10;
+ }
+ }
+
+ message List {
+ repeated ArgValue element = 1;
+ }
+ message Map {
+ map<string, ArgValue> map = 1;
+ }
+ message RemoteObject {
+ string id = 1;
+ }
+
+ message ConstructObject {
+ string className = 1;
+ repeated ArgValue values = 2;
+ }
+
+ message DestructObject {
+ RemoteObject remote_object = 1;
+ }
+
+ message CallMethod {
+ RemoteObject remote_object = 1;
+ string method_name = 2;
+ repeated ArgValue arg_values = 3;
+ }
+
+ message CallFunction {
+ string module_name = 1;
+ repeated ArgValue arg_values = 3;
+ }
+}
+
+
+message MlCommand {
Review Comment:
Note:
The PR is WIP,
and I will remove the whole `MlCommand` message definition in following commits, we only need the `RemoteCall` message for ML cases.
--
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.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] WeichenXu123 commented on a diff in pull request #39985: [SPARK-42412][WIP] Initial prototype implementation of PySpark ML via Spark connect
Posted by "WeichenXu123 (via GitHub)" <gi...@apache.org>.
WeichenXu123 commented on code in PR #39985:
URL: https://github.com/apache/spark/pull/39985#discussion_r1107912258
##########
connector/connect/common/src/main/protobuf/spark/connect/relations.proto:
##########
@@ -76,6 +76,8 @@ message Relation {
StatFreqItems freq_items = 106;
StatSampleBy sample_by = 107;
+ int64 server_side_dataframe_id = 108;
Review Comment:
@amaliujia
> This should be in RelationCommon?
No. Because this might be for either DataFrame object or other type objects (e.g. Model type object)
--
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.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] WeichenXu123 commented on a diff in pull request #39985: [SPARK-42412][WIP] Initial prototype implementation of PySpark ML via Spark connect
Posted by "WeichenXu123 (via GitHub)" <gi...@apache.org>.
WeichenXu123 commented on code in PR #39985:
URL: https://github.com/apache/spark/pull/39985#discussion_r1107913325
##########
connector/connect/server/pom.xml:
##########
@@ -93,6 +93,18 @@
</exclusion>
</exclusions>
</dependency>
+ <dependency>
+ <groupId>org.apache.spark</groupId>
+ <artifactId>spark-mllib_${scala.binary.version}</artifactId>
+ <version>${project.version}</version>
+ <scope>provided</scope>
+ <exclusions>
+ <exclusion>
+ <groupId>com.google.guava</groupId>
+ <artifactId>guava</artifactId>
+ </exclusion>
+ </exclusions>
+ </dependency>
Review Comment:
We need to make it depends on mllib project
--
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.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] zhengruifeng commented on a diff in pull request #39985: [SPARK-42412][WIP] Initial prototype implementation of PySpark ML via Spark connect
Posted by "zhengruifeng (via GitHub)" <gi...@apache.org>.
zhengruifeng commented on code in PR #39985:
URL: https://github.com/apache/spark/pull/39985#discussion_r1108192576
##########
connector/connect/common/src/main/protobuf/spark/connect/relations.proto:
##########
@@ -76,6 +76,8 @@ message Relation {
StatFreqItems freq_items = 106;
StatSampleBy sample_by = 107;
+ int64 server_side_dataframe_id = 108;
Review Comment:
what is the dataframe/logicalplan here? is it the output of model prediction (`model.transform`) ?
--
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.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] WeichenXu123 commented on a diff in pull request #39985: [SPARK-42412][WIP] Initial prototype implementation of PySpark ML via Spark connect
Posted by "WeichenXu123 (via GitHub)" <gi...@apache.org>.
WeichenXu123 commented on code in PR #39985:
URL: https://github.com/apache/spark/pull/39985#discussion_r1104456256
##########
connector/connect/common/src/main/protobuf/spark/connect/ml.proto:
##########
@@ -0,0 +1,165 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+syntax = 'proto3';
+
+package spark.connect;
+
+import "spark/connect/relations.proto";
+
+option java_multiple_files = true;
+option java_package = "org.apache.spark.connect.proto";
+
+
+message RemoteCall {
+ oneof call_type {
+ ConstructObject construct_object = 1;
+ DestructObject destruct_object = 2;
+ CallMethod call_method = 3;
+ CallFunction call_function = 4;
+ }
+
+ message ArgValue {
+ oneof arg_type {
+ int32 int32_value = 1;
+ int64 int64_value = 2;
+ float float_value = 3;
+ double double_value = 4;
+ bool bool_value = 5;
+ string string_value = 6;
+ List list = 7;
+ Map map = 8;
+ Relation relation = 9;
+ RemoteObject remote_object = 10;
Review Comment:
Note: I will add `Matrix` and `Vector` message type field into oneof field of `ArgValue` later. Some ML methods return matrix or vector objects.
--
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.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] amaliujia commented on a diff in pull request #39985: [SPARK-42412][WIP] Initial prototype implementation of PySpark ML via Spark connect
Posted by "amaliujia (via GitHub)" <gi...@apache.org>.
amaliujia commented on code in PR #39985:
URL: https://github.com/apache/spark/pull/39985#discussion_r1107691911
##########
connector/connect/common/src/main/protobuf/spark/connect/relations.proto:
##########
@@ -76,6 +76,8 @@ message Relation {
StatFreqItems freq_items = 106;
StatSampleBy sample_by = 107;
+ int64 server_side_dataframe_id = 108;
Review Comment:
It makes sense to have its own message if it is a rel type.
--
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.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] WeichenXu123 commented on a diff in pull request #39985: [SPARK-42412][WIP] Initial prototype implementation of PySpark ML via Spark connect
Posted by "WeichenXu123 (via GitHub)" <gi...@apache.org>.
WeichenXu123 commented on code in PR #39985:
URL: https://github.com/apache/spark/pull/39985#discussion_r1107913325
##########
connector/connect/server/pom.xml:
##########
@@ -93,6 +93,18 @@
</exclusion>
</exclusions>
</dependency>
+ <dependency>
+ <groupId>org.apache.spark</groupId>
+ <artifactId>spark-mllib_${scala.binary.version}</artifactId>
+ <version>${project.version}</version>
+ <scope>provided</scope>
+ <exclusions>
+ <exclusion>
+ <groupId>com.google.guava</groupId>
+ <artifactId>guava</artifactId>
+ </exclusion>
+ </exclusions>
+ </dependency>
Review Comment:
We need to make it depends on mllib-local project
--
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.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] WeichenXu123 commented on a diff in pull request #39985: [SPARK-42412][WIP] Initial prototype implementation of PySpark ML via Spark connect
Posted by "WeichenXu123 (via GitHub)" <gi...@apache.org>.
WeichenXu123 commented on code in PR #39985:
URL: https://github.com/apache/spark/pull/39985#discussion_r1108234304
##########
connector/connect/common/src/main/protobuf/spark/connect/relations.proto:
##########
@@ -76,6 +76,8 @@ message Relation {
StatFreqItems freq_items = 106;
StatSampleBy sample_by = 107;
+ int64 server_side_dataframe_id = 108;
Review Comment:
> is it the output of model prediction (model.transform)
Yes. This is one case.
Other cases are: Some model attributes and model summary methods might return spark DataFrame.
--
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.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] WeichenXu123 commented on a diff in pull request #39985: [SPARK-42412][WIP] Initial prototype implementation of PySpark ML via Spark connect
Posted by "WeichenXu123 (via GitHub)" <gi...@apache.org>.
WeichenXu123 commented on code in PR #39985:
URL: https://github.com/apache/spark/pull/39985#discussion_r1108137184
##########
connector/connect/common/src/main/protobuf/spark/connect/relations.proto:
##########
@@ -76,6 +76,8 @@ message Relation {
StatFreqItems freq_items = 106;
StatSampleBy sample_by = 107;
+ int64 server_side_dataframe_id = 108;
Review Comment:
@amaliujia
You can see this line:
https://github.com/apache/spark/blob/a7a5e30a2db8561ebd51f90e3690ac5f75c6ae60/connector/connect/server/src/main/scala/org/apache/spark/sql/connect/planner/SparkConnectPlanner.scala#L119
```
class SparkConnectPlanner(val sessionHolder: SessionHolder) {
...
def transformRelation(rel: proto.Relation): LogicalPlan = {
rel.getRelTypeCase match {
...
case proto.Relation.RelTypeCase.SERVER_SIDE_DATAFRAME_ID =>
sessionHolder.serverSideObjectManager.getObject(rel.getServerSideDataframeId)
.asInstanceOf[Dataset[Any]].logicalPlan
...
}
```
i.e., it means we get the spark dataframe plan via query the id from `sessionHolder.serverSideObjectManager`
--
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.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] WeichenXu123 commented on a diff in pull request #39985: [SPARK-42412][WIP] Initial prototype implementation of PySpark ML via Spark connect
Posted by "WeichenXu123 (via GitHub)" <gi...@apache.org>.
WeichenXu123 commented on code in PR #39985:
URL: https://github.com/apache/spark/pull/39985#discussion_r1107183606
##########
connector/connect/common/src/main/protobuf/spark/connect/ml.proto:
##########
@@ -0,0 +1,166 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+syntax = 'proto3';
+
+package spark.connect;
+
+import "spark/connect/relations.proto";
+
+option java_multiple_files = true;
+option java_package = "org.apache.spark.connect.proto";
+
+
+message RemoteCall {
+ oneof call_type {
+ ConstructObject construct_object = 1;
+ DestructObject destruct_object = 2;
+ CallMethod call_method = 3;
+ CallFunction call_function = 4;
+ }
Review Comment:
@grundprinzip
PR is updated.
The first stage of this PR is complete, it can support all java-backend estimator's fit / transform / all model attributes / model summary.
I haven't tested pyspark Pipeline, but I guess it is already supported directly without other change.
TODO:
* Support pyspark CrossValidator and other meta-estimator
* Support pyspark evaluator
* remote object GC
* Some minor fixes (e.g. set model params / model uid correctly)
* add test.
I will attach design doc later.
--
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.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] WeichenXu123 commented on a diff in pull request #39985: [SPARK-42412][WIP] Initial prototype implementation of PySpark ML via Spark connect
Posted by "WeichenXu123 (via GitHub)" <gi...@apache.org>.
WeichenXu123 commented on code in PR #39985:
URL: https://github.com/apache/spark/pull/39985#discussion_r1107183606
##########
connector/connect/common/src/main/protobuf/spark/connect/ml.proto:
##########
@@ -0,0 +1,166 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+syntax = 'proto3';
+
+package spark.connect;
+
+import "spark/connect/relations.proto";
+
+option java_multiple_files = true;
+option java_package = "org.apache.spark.connect.proto";
+
+
+message RemoteCall {
+ oneof call_type {
+ ConstructObject construct_object = 1;
+ DestructObject destruct_object = 2;
+ CallMethod call_method = 3;
+ CallFunction call_function = 4;
+ }
Review Comment:
@grundprinzip
PR is updated.
The first stage of this PR is complete, it can support all java-backend estimator's fit / transform / all model attributes / model summary.
I haven't tested pyspark Pipeline, but I guess it is already supported directly without other change.
TODO:
* Support pyspark CrossValidator and other meta-estimator
* Support pyspark evaluator
* remote object GC
* Some minor fixes (e.g. set model params / model uid correctly)
I will attach design doc later.
--
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.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] WeichenXu123 commented on a diff in pull request #39985: [SPARK-42412][WIP] Initial prototype implementation of PySpark ML via Spark connect
Posted by "WeichenXu123 (via GitHub)" <gi...@apache.org>.
WeichenXu123 commented on code in PR #39985:
URL: https://github.com/apache/spark/pull/39985#discussion_r1107893116
##########
connector/connect/common/src/main/protobuf/spark/connect/ml.proto:
##########
@@ -0,0 +1,166 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+syntax = 'proto3';
+
+package spark.connect;
+
+import "spark/connect/relations.proto";
+
+option java_multiple_files = true;
+option java_package = "org.apache.spark.connect.proto";
+
+
+message RemoteCall {
+ oneof call_type {
+ ConstructObject construct_object = 1;
+ DestructObject destruct_object = 2;
+ CallMethod call_method = 3;
+ CallFunction call_function = 4;
+ }
Review Comment:
sure! I will share design doc later for current design intention explanation and list other interface options for discussion.
--
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.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] amaliujia commented on a diff in pull request #39985: [SPARK-42412][WIP] Initial prototype implementation of PySpark ML via Spark connect
Posted by "amaliujia (via GitHub)" <gi...@apache.org>.
amaliujia commented on code in PR #39985:
URL: https://github.com/apache/spark/pull/39985#discussion_r1107687401
##########
connector/connect/server/src/main/scala/org/apache/spark/sql/connect/planner/MLUtils.scala:
##########
@@ -0,0 +1,79 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.spark.sql.connect.planner
+
+import org.apache.spark.ml.param.{Param, Params}
+
+object SparkConnectUtils {
+
+ private def castParamValue(paramType: Class[_], paramValue: Object): Object = {
+ if (paramType == classOf[Int]) {
+ java.lang.Integer.valueOf(paramValue.asInstanceOf[java.lang.Number].intValue())
+ } else if (paramType == classOf[Long]) {
+ java.lang.Long.valueOf(paramValue.asInstanceOf[java.lang.Number].longValue())
+ } else if (paramType == classOf[Float]) {
+ java.lang.Float.valueOf(paramValue.asInstanceOf[java.lang.Number].floatValue())
+ } else if (paramType == classOf[Double]) {
+ java.lang.Double.valueOf(paramValue.asInstanceOf[java.lang.Number].doubleValue())
+ } else if (paramType == classOf[Array[Int]]) {
+ paramValue.asInstanceOf[List[Object]]
+ .map(castParamValue(classOf[Int], _).asInstanceOf[java.lang.Integer])
+ .map(_.intValue()).toArray
+ } else if (paramType == classOf[Array[Long]]) {
+ paramValue.asInstanceOf[List[Object]]
+ .map(castParamValue(classOf[Long], _).asInstanceOf[java.lang.Long])
+ .map(_.longValue()).toArray
+ } else if (paramType == classOf[Array[Float]]) {
+ paramValue.asInstanceOf[List[Object]]
+ .map(castParamValue(classOf[Float], _).asInstanceOf[java.lang.Float])
+ .map(_.floatValue()).toArray
+ } else if (paramType == classOf[Array[Double]]) {
+ paramValue.asInstanceOf[List[Object]]
+ .map(castParamValue(classOf[Double], _).asInstanceOf[java.lang.Double])
+ .map(_.doubleValue()).toArray
+ } else if (paramType == classOf[Array[Boolean]]) {
+ paramValue.asInstanceOf[List[java.lang.Boolean]].map(_.booleanValue()).toArray
+ } else if (paramType == classOf[Array[String]]) {
+ paramValue.asInstanceOf[List[String]].toArray
+ } else {
+ throw new RuntimeException()
+ }
+ }
+
+ private def makeParamPair(
+ instance: Params, paramName: String, paramValue: Object
Review Comment:
```
./build/mvn -Pscala-2.12 scalafmt:format -Dscalafmt.skip=false -Dscalafmt.validateOnly=false -Dscalafmt.changedOnly=false -pl connector/connect/common -pl connector/connect/server
```
You can run this command to format the server side code.
--
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.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] amaliujia commented on a diff in pull request #39985: [SPARK-42412][WIP] Initial prototype implementation of PySpark ML via Spark connect
Posted by "amaliujia (via GitHub)" <gi...@apache.org>.
amaliujia commented on code in PR #39985:
URL: https://github.com/apache/spark/pull/39985#discussion_r1107687401
##########
connector/connect/server/src/main/scala/org/apache/spark/sql/connect/planner/MLUtils.scala:
##########
@@ -0,0 +1,79 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.spark.sql.connect.planner
+
+import org.apache.spark.ml.param.{Param, Params}
+
+object SparkConnectUtils {
+
+ private def castParamValue(paramType: Class[_], paramValue: Object): Object = {
+ if (paramType == classOf[Int]) {
+ java.lang.Integer.valueOf(paramValue.asInstanceOf[java.lang.Number].intValue())
+ } else if (paramType == classOf[Long]) {
+ java.lang.Long.valueOf(paramValue.asInstanceOf[java.lang.Number].longValue())
+ } else if (paramType == classOf[Float]) {
+ java.lang.Float.valueOf(paramValue.asInstanceOf[java.lang.Number].floatValue())
+ } else if (paramType == classOf[Double]) {
+ java.lang.Double.valueOf(paramValue.asInstanceOf[java.lang.Number].doubleValue())
+ } else if (paramType == classOf[Array[Int]]) {
+ paramValue.asInstanceOf[List[Object]]
+ .map(castParamValue(classOf[Int], _).asInstanceOf[java.lang.Integer])
+ .map(_.intValue()).toArray
+ } else if (paramType == classOf[Array[Long]]) {
+ paramValue.asInstanceOf[List[Object]]
+ .map(castParamValue(classOf[Long], _).asInstanceOf[java.lang.Long])
+ .map(_.longValue()).toArray
+ } else if (paramType == classOf[Array[Float]]) {
+ paramValue.asInstanceOf[List[Object]]
+ .map(castParamValue(classOf[Float], _).asInstanceOf[java.lang.Float])
+ .map(_.floatValue()).toArray
+ } else if (paramType == classOf[Array[Double]]) {
+ paramValue.asInstanceOf[List[Object]]
+ .map(castParamValue(classOf[Double], _).asInstanceOf[java.lang.Double])
+ .map(_.doubleValue()).toArray
+ } else if (paramType == classOf[Array[Boolean]]) {
+ paramValue.asInstanceOf[List[java.lang.Boolean]].map(_.booleanValue()).toArray
+ } else if (paramType == classOf[Array[String]]) {
+ paramValue.asInstanceOf[List[String]].toArray
+ } else {
+ throw new RuntimeException()
+ }
+ }
+
+ private def makeParamPair(
+ instance: Params, paramName: String, paramValue: Object
Review Comment:
```
./build/mvn -Pscala-2.12 scalafmt:format -Dscalafmt.skip=false -Dscalafmt.validateOnly=false -Dscalafmt.changedOnly=false -pl connector/connect/common -pl connector/connect/server
```
You can run this command to format the common/server side code.
--
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.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] grundprinzip commented on a diff in pull request #39985: [SPARK-42412][WIP] Initial prototype implementation of PySpark ML via Spark connect
Posted by "grundprinzip (via GitHub)" <gi...@apache.org>.
grundprinzip commented on code in PR #39985:
URL: https://github.com/apache/spark/pull/39985#discussion_r1107688477
##########
connector/connect/common/src/main/protobuf/spark/connect/relations.proto:
##########
@@ -76,6 +76,8 @@ message Relation {
StatFreqItems freq_items = 106;
StatSampleBy sample_by = 107;
+ int64 server_side_dataframe_id = 108;
Review Comment:
This should be it's own proper message.
```
ServerSideDataFrame server_side_data_frame = 108;
message ServerSideDataFrame {
int64 id = 0;
}
```
--
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.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] WeichenXu123 commented on a diff in pull request #39985: [SPARK-42412][WIP] Initial prototype implementation of PySpark ML via Spark connect
Posted by "WeichenXu123 (via GitHub)" <gi...@apache.org>.
WeichenXu123 commented on code in PR #39985:
URL: https://github.com/apache/spark/pull/39985#discussion_r1106969521
##########
connector/connect/server/src/main/scala/org/apache/spark/sql/connect/planner/MLUtils.scala:
##########
@@ -0,0 +1,79 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.spark.sql.connect.planner
+
+import org.apache.spark.ml.param.{Param, Params}
+
+object SparkConnectUtils {
+
+ private def castParamValue(paramType: Class[_], paramValue: Object): Object = {
+ if (paramType == classOf[Int]) {
Review Comment:
Unfortunately, we can't .
--
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.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] amaliujia commented on a diff in pull request #39985: [SPARK-42412][WIP] Initial prototype implementation of PySpark ML via Spark connect
Posted by "amaliujia (via GitHub)" <gi...@apache.org>.
amaliujia commented on code in PR #39985:
URL: https://github.com/apache/spark/pull/39985#discussion_r1107686641
##########
connector/connect/server/pom.xml:
##########
@@ -93,6 +93,18 @@
</exclusion>
</exclusions>
</dependency>
+ <dependency>
+ <groupId>org.apache.spark</groupId>
+ <artifactId>spark-mllib_${scala.binary.version}</artifactId>
+ <version>${project.version}</version>
+ <scope>provided</scope>
+ <exclusions>
+ <exclusion>
+ <groupId>com.google.guava</groupId>
+ <artifactId>guava</artifactId>
+ </exclusion>
+ </exclusions>
+ </dependency>
Review Comment:
cc @LuciferYang pom.xml is touched here.
--
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.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] WeichenXu123 commented on a diff in pull request #39985: [SPARK-42412][WIP] Initial prototype implementation of PySpark ML via Spark connect
Posted by "WeichenXu123 (via GitHub)" <gi...@apache.org>.
WeichenXu123 commented on code in PR #39985:
URL: https://github.com/apache/spark/pull/39985#discussion_r1107183606
##########
connector/connect/common/src/main/protobuf/spark/connect/ml.proto:
##########
@@ -0,0 +1,166 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+syntax = 'proto3';
+
+package spark.connect;
+
+import "spark/connect/relations.proto";
+
+option java_multiple_files = true;
+option java_package = "org.apache.spark.connect.proto";
+
+
+message RemoteCall {
+ oneof call_type {
+ ConstructObject construct_object = 1;
+ DestructObject destruct_object = 2;
+ CallMethod call_method = 3;
+ CallFunction call_function = 4;
+ }
Review Comment:
@grundprinzip
PR is updated.
The first stage of this PR is complete, it can support all java-backend estimator's fit / transform / all model attributes / model summary / model saving loading.
I haven't tested pyspark Pipeline, but I guess it is already supported directly without other change.
You can test it by the code pasted in PR description.
TODO list see my PR description.
I will attach design doc later.
The final goal is to seemingly support all pyspark ML estimators / transformers without any user code changes and keep the same behavior with non-spark-connect mode. With current design, we are not far from the goal and we don't need to change too much code.
--
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.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] WeichenXu123 commented on a diff in pull request #39985: [SPARK-42412][WIP] Initial prototype implementation of PySpark ML via Spark connect
Posted by "WeichenXu123 (via GitHub)" <gi...@apache.org>.
WeichenXu123 commented on code in PR #39985:
URL: https://github.com/apache/spark/pull/39985#discussion_r1107183606
##########
connector/connect/common/src/main/protobuf/spark/connect/ml.proto:
##########
@@ -0,0 +1,166 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+syntax = 'proto3';
+
+package spark.connect;
+
+import "spark/connect/relations.proto";
+
+option java_multiple_files = true;
+option java_package = "org.apache.spark.connect.proto";
+
+
+message RemoteCall {
+ oneof call_type {
+ ConstructObject construct_object = 1;
+ DestructObject destruct_object = 2;
+ CallMethod call_method = 3;
+ CallFunction call_function = 4;
+ }
Review Comment:
@grundprinzip
PR is updated.
The first stage of this PR is complete, it can support all java-backend estimator's fit / transform / all model attributes / model summary / model saving loading.
I haven't tested pyspark Pipeline, but I guess it is already supported directly without other change.
You can test it by the code pasted in PR description.
TODO list is included in my PR description.
I will attach design doc later.
The final goal is to seamlessly support all pyspark ML estimators / transformers without any user code changes and keep the same behavior with non-spark-connect mode. With current design, we are not far from the goal and we don't need to change too much code.
--
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.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] WeichenXu123 commented on a diff in pull request #39985: [SPARK-42412][WIP] Initial prototype implementation of PySpark ML via Spark connect
Posted by "WeichenXu123 (via GitHub)" <gi...@apache.org>.
WeichenXu123 commented on code in PR #39985:
URL: https://github.com/apache/spark/pull/39985#discussion_r1107183606
##########
connector/connect/common/src/main/protobuf/spark/connect/ml.proto:
##########
@@ -0,0 +1,166 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+syntax = 'proto3';
+
+package spark.connect;
+
+import "spark/connect/relations.proto";
+
+option java_multiple_files = true;
+option java_package = "org.apache.spark.connect.proto";
+
+
+message RemoteCall {
+ oneof call_type {
+ ConstructObject construct_object = 1;
+ DestructObject destruct_object = 2;
+ CallMethod call_method = 3;
+ CallFunction call_function = 4;
+ }
Review Comment:
@grundprinzip
PR is updated.
The first stage of this PR is complete, it can support java-backend estimator's fit / transform / all model attributes / model summary.
I haven't tested pyspark Pipeline, but I guess it is already supported directly without other change.
TODO:
* Support pyspark CrossValidator and other meta-estimator
* Support pyspark evaluator
* remote object GC
* Some minor fixes (e.g. set model params / model uid correctly)
I will attach design doc later.
--
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.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] grundprinzip commented on a diff in pull request #39985: [SPARK-42412][WIP] Initial prototype implementation of PySpark ML via Spark connect
Posted by "grundprinzip (via GitHub)" <gi...@apache.org>.
grundprinzip commented on code in PR #39985:
URL: https://github.com/apache/spark/pull/39985#discussion_r1107687377
##########
connector/connect/common/src/main/protobuf/spark/connect/ml.proto:
##########
@@ -0,0 +1,166 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+syntax = 'proto3';
+
+package spark.connect;
+
+import "spark/connect/relations.proto";
+
+option java_multiple_files = true;
+option java_package = "org.apache.spark.connect.proto";
+
+
+message RemoteCall {
+ oneof call_type {
+ ConstructObject construct_object = 1;
+ DestructObject destruct_object = 2;
+ CallMethod call_method = 3;
+ CallFunction call_function = 4;
+ }
+
+ message ArgValue {
Review Comment:
You dont need to create a full expression, just use the `Expression.Literal` message directly.
--
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.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] amaliujia commented on a diff in pull request #39985: [SPARK-42412][WIP] Initial prototype implementation of PySpark ML via Spark connect
Posted by "amaliujia (via GitHub)" <gi...@apache.org>.
amaliujia commented on code in PR #39985:
URL: https://github.com/apache/spark/pull/39985#discussion_r1107686002
##########
connector/connect/common/src/main/protobuf/spark/connect/relations.proto:
##########
@@ -76,6 +76,8 @@ message Relation {
StatFreqItems freq_items = 106;
StatSampleBy sample_by = 107;
+ int64 server_side_dataframe_id = 108;
Review Comment:
This should be in `RelationCommon`?
--
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.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] WeichenXu123 commented on a diff in pull request #39985: [SPARK-42412][WIP] Initial prototype implementation of PySpark ML via Spark connect
Posted by "WeichenXu123 (via GitHub)" <gi...@apache.org>.
WeichenXu123 commented on code in PR #39985:
URL: https://github.com/apache/spark/pull/39985#discussion_r1107922572
##########
connector/connect/common/src/main/protobuf/spark/connect/relations.proto:
##########
@@ -76,6 +76,8 @@ message Relation {
StatFreqItems freq_items = 106;
StatSampleBy sample_by = 107;
+ int64 server_side_dataframe_id = 108;
Review Comment:
Can we use `RelationCommon` ? I am not quite familiar with Relation part,
the server side dataframe might contain some stage (e.g. using RDD api) that Spark connect proto does not support
--
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.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] amaliujia commented on a diff in pull request #39985: [SPARK-42412][WIP] Initial prototype implementation of PySpark ML via Spark connect
Posted by "amaliujia (via GitHub)" <gi...@apache.org>.
amaliujia commented on code in PR #39985:
URL: https://github.com/apache/spark/pull/39985#discussion_r1107930496
##########
connector/connect/common/src/main/protobuf/spark/connect/relations.proto:
##########
@@ -76,6 +76,8 @@ message Relation {
StatFreqItems freq_items = 106;
StatSampleBy sample_by = 107;
+ int64 server_side_dataframe_id = 108;
Review Comment:
The problem here is that I do not fully understand the semantics of this `ServerSideDataFrame`. If you can help us understand what exactly `ServerSideDataFrame` will do, probably we can help to see how does it better fit in the Connect proto.
--
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.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] WeichenXu123 commented on a diff in pull request #39985: [SPARK-42412][WIP] Initial prototype implementation of PySpark ML via Spark connect
Posted by "WeichenXu123 (via GitHub)" <gi...@apache.org>.
WeichenXu123 commented on code in PR #39985:
URL: https://github.com/apache/spark/pull/39985#discussion_r1106972507
##########
connector/connect/server/src/main/scala/org/apache/spark/sql/connect/planner/SparkMLPlanner.scala:
##########
@@ -0,0 +1,402 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.sql.connect.planner
+
+import io.grpc.stub.StreamObserver
+import java.lang
+import scala.reflect.runtime.universe
+
+import org.apache.spark.connect.proto
+import org.apache.spark.connect.proto.{ExecutePlanRequest, ExecutePlanResponse, RemoteCall}
+import org.apache.spark.sql.{DataFrame, Dataset}
+import org.apache.spark.sql.connect.service.{SessionHolder}
+import org.apache.spark.util.Utils
+
+object SparkMLPlanner {
+
+ val runtimeMirror = universe.runtimeMirror(Utils.getContextOrSparkClassLoader)
+
+ def buildRemoteCallResponse(
+ request: ExecutePlanRequest,
+ responseObserver: StreamObserver[ExecutePlanResponse],
+ returnValueOpt: Option[RemoteCall.ArgValue]
+ ): Unit = {
+ val response = proto.ExecutePlanResponse.newBuilder
+ .setClientId(request.getClientId)
+
+ returnValueOpt.foreach(response.setRemoteCallReturnValue(_))
+
+ responseObserver.onNext(response.build())
+ responseObserver.onCompleted()
+ }
+
+ def loadInputRelation(sessionHolder: SessionHolder, inputRelation: proto.Relation): DataFrame = {
+ val relationalPlanner = new SparkConnectPlanner(sessionHolder)
+ val plan = relationalPlanner.transformRelation(inputRelation)
+ Dataset.ofRows(sessionHolder.session, plan)
+ }
+
+ /*
+ def createStage(uid: String, className: String): PipelineStage = {
+ val clazz = Utils.classForName(className)
+ val ctorOpt = clazz.getConstructors.find { ctor =>
+ // Find the constructor with signature `this(uid: String)`
+ ctor.getParameterCount == 1 && ctor.getParameterTypes()(0).eq(classOf[String])
+ }
+ if (ctorOpt.isEmpty) {
+ throw new RuntimeException(
+ s"Could not find 'this(uid: String)' constructor for class $className"
+ )
+ }
+
+ ctorOpt.get.newInstance(uid).asInstanceOf[PipelineStage]
+ }
+
+ def extractParamPair(
+ instance: Params,
+ paramName: String,
+ protoValue: MlCommand.ParamValue
+ ): (Param[Any], Any) = {
+ import MlCommand.ParamValue.ValueCase
+
+ val param = instance.getParam(paramName)
+ val valueType = param.paramValueClassTag.runtimeClass
+ val extractedValue = protoValue.getValueCase match {
+ case ValueCase.INT_VAL =>
+ assert(valueType == classOf[Int])
+ protoValue.getIntVal
+
+ case ValueCase.LONG_VAL =>
+ if (valueType == classOf[Long]) {
+ protoValue.getLongVal
+ } else if (valueType == classOf[Int]) {
+ protoValue.getLongVal.toInt
+ } else {
+ throw new java.lang.AssertionError()
+ }
+
+ case ValueCase.FLOAT_VAL =>
+ assert(valueType == classOf[Float])
+ protoValue.getFloatVal
+
+ case ValueCase.DOUBLE_VAL =>
+ if (valueType == classOf[Double]) {
+ protoValue.getDoubleVal
+ } else if (valueType == classOf[Float]) {
+ protoValue.getDoubleVal.toFloat
+ } else {
+ throw new java.lang.AssertionError()
+ }
+
+ case ValueCase.BOOL_VAL =>
+ assert(valueType == classOf[Boolean])
+ protoValue.getBoolVal
+
+ case ValueCase.STR_VAL =>
+ assert(valueType == classOf[String])
+ protoValue.getStrVal
+
+ case ValueCase.INT_ARRAY =>
+ assert(valueType == classOf[Array[Int]])
+ protoValue.getIntArray.getElementList.stream().mapToInt(_.toInt).toArray
+
+ case ValueCase.LONG_ARRAY =>
+ val elemList = protoValue.getLongArray.getElementList
+ if (valueType == classOf[Array[Long]]) {
+ elemList.stream().mapToLong(_.toLong).toArray
+ } else if (valueType == classOf[Array[Int]]) {
+ elemList.stream().mapToInt(_.toInt).toArray
+ } else {
+ throw new java.lang.AssertionError()
+ }
+
+ case ValueCase.FLOAT_ARRAY =>
+ assert(valueType == classOf[Float])
+ val floatList = protoValue.getFloatArray.getElementList
+ val floatArray = new Array[Float](floatList.size())
+ for (i <- 0 until floatList.size()) {
+ floatArray(i) = floatList.get(i)
+ }
+ floatArray
+
+ case ValueCase.DOUBLE_ARRAY =>
+ val doubleList = protoValue.getDoubleArray.getElementList
+ if (valueType == classOf[Array[Double]]) {
+ doubleList.stream().mapToDouble(_.toDouble).toArray
+ } else if (valueType == classOf[Array[Float]]) {
+ val floatArray = new Array[Float](doubleList.size())
+ for (i <- 0 until doubleList.size()) {
+ floatArray(i) = doubleList.get(i).toFloat
+ }
+ floatArray
+ } else {
+ throw new java.lang.AssertionError()
+ }
+
+ case ValueCase.STR_ARRAY =>
+ assert(valueType == classOf[Array[String]])
+ protoValue.getStrArray.getElementList.toArray(Array[String]())
+
+ case ValueCase.BOOL_ARRAY =>
+ assert(valueType == classOf[Array[Boolean]])
+ val boolList = protoValue.getBoolArray.getElementList
+ val boolArray = new Array[Boolean](boolList.size())
+ for (i <- 0 until boolList.size()) {
+ boolArray(i) = boolList.get(i)
+ }
+ boolArray
+
+ case _ =>
+ throw InvalidPlanInput()
+ }
+ (param, extractedValue)
+ }
+
+ def setParams(instance: Params, protoParams: MlCommand.Params): Unit = {
+ protoParams.getParamsMap.forEach { (paramName, protoValue) =>
+ val (param, paramValue) = extractParamPair(
+ instance, paramName, protoValue
+ )
+ instance.set(param, paramValue)
+ }
+ protoParams.getDefaultParamsMap.forEach { (paramName, protoValue) =>
+ val (param, paramValue) = extractParamPair(
+ instance, paramName, protoValue
+ )
+ instance._setDefault(param, paramValue)
+ }
+ }
+
+ def handleRemoteCall(
+ sessionHolder: SessionHolder,
+ request: ExecutePlanRequest,
+ responseObserver: StreamObserver[ExecutePlanResponse]
+ ): Unit = {
+ val mlCommand = request.getPlan.getMlCommand
+ mlCommand.getOpTypeCase match {
+ case proto.MlCommand.OpTypeCase.CONSTRUCT_STAGE =>
+ val constructStage = mlCommand.getConstructStage
+ val stage = createStage(
+ constructStage.getUid,
+ constructStage.getClassName
+ )
+ val objectId = serverSideObjectManager.registerObject(stage)
+
+ val resp = ExecutePlanResponse.MlCommandResponse
+ .newBuilder.setServerSideObjectId(objectId).build()
+
+ buildMlResponse(request, responseObserver, Some(resp))
+
+ case proto.MlCommand.OpTypeCase.DESTRUCT_OBJECT =>
+ serverSideObjectManager.removeObject(mlCommand.getDestructObject.getId)
+
+ case proto.MlCommand.OpTypeCase.FIT =>
+ val estimator = serverSideObjectManager
+ .getObject(mlCommand.getFit.getId).asInstanceOf[Estimator[_]]
+ val inputDF = loadInputRelation(session, mlCommand.getFit.getInput)
+ val model = estimator.fit(inputDF)
+ val modelObjId = serverSideObjectManager.
+ registerObject(model.asInstanceOf[Object])
+
+ val resp = ExecutePlanResponse.MlCommandResponse
+ .newBuilder.setServerSideObjectId(modelObjId).build()
+
+ buildMlResponse(request, responseObserver, Some(resp))
+
+ case proto.MlCommand.OpTypeCase.TRANSFORM =>
+ val transformer = serverSideObjectManager.getObject(mlCommand.getTransform.getId)
+ .asInstanceOf[Transformer]
+ val inputDF = loadInputRelation(session, mlCommand.getTransform.getInput)
+ val transformedDF = transformer.transform(inputDF)
+ val transformedDFId = serverSideObjectManager.registerObject(transformedDF)
+
+ val resp = ExecutePlanResponse.MlCommandResponse
+ .newBuilder.setServerSideObjectId(transformedDFId).build()
+
+ buildMlResponse(request, responseObserver, Some(resp))
+
+ case proto.MlCommand.OpTypeCase.TRANSFER_PARAMS_TO_SERVER =>
+ val instance = serverSideObjectManager.getObject(mlCommand.getTransferParamsToServer.getId)
+ .asInstanceOf[Params]
+ val protoParams = mlCommand.getTransferParamsToServer.getParams
+ setParams(instance, protoParams)
+
+ buildMlResponse(request, responseObserver, None)
+
+ case _ =>
+ throw new UnsupportedOperationException(s"${mlCommand.getOpTypeCase} not supported.")
+ }
+ }
+ */
+
+ def parseArg(protoArg: proto.RemoteCall.ArgValue, sessionHolder: SessionHolder): Object = {
+ import scala.collection.JavaConverters._
+ import proto.RemoteCall.ArgValue
+ protoArg.getArgTypeCase match {
+ case ArgValue.ArgTypeCase.INT32_VALUE =>
+ lang.Integer.valueOf(protoArg.getInt32Value)
+ case ArgValue.ArgTypeCase.INT64_VALUE =>
+ lang.Long.valueOf(protoArg.getInt64Value)
+ case ArgValue.ArgTypeCase.FLOAT_VALUE =>
+ lang.Float.valueOf(protoArg.getFloatValue)
+ case ArgValue.ArgTypeCase.DOUBLE_VALUE =>
+ lang.Double.valueOf(protoArg.getDoubleValue)
+ case ArgValue.ArgTypeCase.STRING_VALUE =>
+ protoArg.getStringValue
+ case ArgValue.ArgTypeCase.BOOL_VALUE =>
+ lang.Boolean.valueOf(protoArg.getBoolValue)
+ case ArgValue.ArgTypeCase.RELATION =>
+ loadInputRelation(sessionHolder, protoArg.getRelation)
+ case ArgValue.ArgTypeCase.REMOTE_OBJECT =>
+ sessionHolder.serverSideObjectManager.getObject(protoArg.getRemoteObject.getId)
+ case ArgValue.ArgTypeCase.LIST =>
+ val protoValues = protoArg.getList.getElementList.asScala.toList
+ protoValues.map { x => parseArg(x, sessionHolder) }
+ case ArgValue.ArgTypeCase.MAP =>
+ val protoMap = protoArg.getMap.getMapMap().asScala.toMap
+ protoMap.mapValues(x => parseArg(x, sessionHolder))
+ case _ =>
+ throw InvalidPlanInput()
+ }
+ }
+
+ /**
+ * Parse prototype message of argument list, returns a tuple of (values, types)
+ */
+ def parseArgs(
+ protoArgList: List[proto.RemoteCall.ArgValue],
+ sessionHolder: SessionHolder
+ ): Array[Object] = {
+ val argValues = new Array[Object](protoArgList.size)
+
+ for (i <- 0 until protoArgList.size) {
+ val protoValue = protoArgList(i)
+ argValues(i) = parseArg(protoValue, sessionHolder)
+ }
+ argValues
+ }
+
+ def serializeValue(value: Object, sessionHolder: SessionHolder): proto.RemoteCall.ArgValue = {
+ val protoBuilder = proto.RemoteCall.ArgValue.newBuilder()
+ value match {
+ case v: lang.Integer =>
+ protoBuilder.setInt32Value(v.intValue())
+ case v: lang.Long =>
+ protoBuilder.setInt64Value(v.longValue())
+ case v: lang.Float =>
+ protoBuilder.setFloatValue(v.floatValue())
+ case v: lang.Double =>
+ protoBuilder.setDoubleValue(v.doubleValue())
+ case v: String =>
+ protoBuilder.setStringValue(v)
+ case v: lang.Boolean =>
+ protoBuilder.setBoolValue(v)
+ case v: List[_] =>
+ // TODO
+ throw new UnsupportedOperationException()
+ case v: Map[_, _] =>
+ // TODO
+ throw new UnsupportedOperationException()
+ case v: Object =>
+ val instanceId = sessionHolder.serverSideObjectManager.registerObject(v)
+ protoBuilder.setRemoteObject(
+ RemoteCall.RemoteObject.newBuilder()
+ .setId(instanceId)
+ .build()
+ )
+ }
+ protoBuilder.build()
+ }
+
+ def invokeMethod(
+ instance: Object,
+ methodName: String,
+ argsProto: List[RemoteCall.ArgValue],
+ sessionHolder: SessionHolder): Option[RemoteCall.ArgValue] = {
+ val argValues = parseArgs(argsProto, sessionHolder)
+ val method = instance.getClass.getMethods().find { method =>
Review Comment:
For methods that might be called frequently, I can cache return result for them.
--
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.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] WeichenXu123 commented on a diff in pull request #39985: [SPARK-42412][WIP] Initial prototype implementation of PySpark ML via Spark connect
Posted by "WeichenXu123 (via GitHub)" <gi...@apache.org>.
WeichenXu123 commented on code in PR #39985:
URL: https://github.com/apache/spark/pull/39985#discussion_r1106651018
##########
connector/connect/common/src/main/protobuf/spark/connect/ml.proto:
##########
@@ -0,0 +1,166 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+syntax = 'proto3';
+
+package spark.connect;
+
+import "spark/connect/relations.proto";
+
+option java_multiple_files = true;
+option java_package = "org.apache.spark.connect.proto";
+
+
+message RemoteCall {
+ oneof call_type {
+ ConstructObject construct_object = 1;
+ DestructObject destruct_object = 2;
+ CallMethod call_method = 3;
+ CallFunction call_function = 4;
+ }
+
+ message ArgValue {
Review Comment:
for array / map expressions, e.g.,`lit([1,2])`, I felt the `expression` is redundant,
```
expr {
unresolved_function {
function_name: "array"
arguments {
literal {
integer: 3
}
}
arguments {
literal {
integer: 4
}
}
}
}
```
when we transfer ML parameters, there might be param with complex array type.
But this is just a prototype code, we can discuss and change it.
##########
connector/connect/common/src/main/protobuf/spark/connect/ml.proto:
##########
@@ -0,0 +1,166 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+syntax = 'proto3';
+
+package spark.connect;
+
+import "spark/connect/relations.proto";
+
+option java_multiple_files = true;
+option java_package = "org.apache.spark.connect.proto";
+
+
+message RemoteCall {
+ oneof call_type {
+ ConstructObject construct_object = 1;
+ DestructObject destruct_object = 2;
+ CallMethod call_method = 3;
+ CallFunction call_function = 4;
+ }
+
+ message ArgValue {
Review Comment:
for array / map expressions, e.g.,`lit([3, 4])`, I felt the `expression` is redundant,
```
expr {
unresolved_function {
function_name: "array"
arguments {
literal {
integer: 3
}
}
arguments {
literal {
integer: 4
}
}
}
}
```
when we transfer ML parameters, there might be param with complex array type.
But this is just a prototype code, we can discuss and change it.
--
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.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] grundprinzip commented on a diff in pull request #39985: [SPARK-42412][WIP] Initial prototype implementation of PySpark ML via Spark connect
Posted by "grundprinzip (via GitHub)" <gi...@apache.org>.
grundprinzip commented on code in PR #39985:
URL: https://github.com/apache/spark/pull/39985#discussion_r1106360916
##########
connector/connect/common/src/main/protobuf/spark/connect/ml.proto:
##########
@@ -0,0 +1,166 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+syntax = 'proto3';
+
+package spark.connect;
+
+import "spark/connect/relations.proto";
+
+option java_multiple_files = true;
+option java_package = "org.apache.spark.connect.proto";
+
+
+message RemoteCall {
+ oneof call_type {
+ ConstructObject construct_object = 1;
+ DestructObject destruct_object = 2;
+ CallMethod call_method = 3;
+ CallFunction call_function = 4;
+ }
+
+ message ArgValue {
Review Comment:
We have expression.Literal for that.
##########
connector/connect/common/src/main/protobuf/spark/connect/ml.proto:
##########
@@ -0,0 +1,166 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+syntax = 'proto3';
+
+package spark.connect;
+
+import "spark/connect/relations.proto";
+
+option java_multiple_files = true;
+option java_package = "org.apache.spark.connect.proto";
+
+
+message RemoteCall {
+ oneof call_type {
+ ConstructObject construct_object = 1;
+ DestructObject destruct_object = 2;
+ CallMethod call_method = 3;
+ CallFunction call_function = 4;
+ }
Review Comment:
You're essentially designing Py4J using Protobuf, but you're no longer expressing the intent of the operation.
Before we can go ahead we need to better idea of what you're designing for and what the declarative interface is.
--
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.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] github-actions[bot] commented on pull request #39985: [SPARK-42412][WIP] Initial prototype implementation of PySpark ML via Spark connect
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on PR #39985:
URL: https://github.com/apache/spark/pull/39985#issuecomment-1603469774
We're closing this PR because it hasn't been updated in a while. This isn't a judgement on the merit of the PR in any way. It's just a way of keeping the PR queue manageable.
If you'd like to revive this PR, please reopen it and ask a committer to remove the Stale tag!
--
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.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org