You are viewing a plain text version of this content. The canonical link for it is here.
Posted to reviews@spark.apache.org by "sven-weber-db (via GitHub)" <gi...@apache.org> on 2026/04/02 16:21:39 UTC
[PR] [WIP][SPARK-56284] Adding UDF worker specification protobuf definition [spark]
sven-weber-db opened a new pull request, #55165:
URL: https://github.com/apache/spark/pull/55165
### What changes were proposed in this pull request?
This PR introduces the protobuf definitions for the UDF worker specification described in [SPIP SPARK-55278](https://issues.apache.org/jira/browse/SPARK-55278) and [this design document](https://docs.google.com/document/d/1Dx9NqHRNuUpatH9DYoFF9cmvUl2fqHT4Rjbyw4EGLHs/edit?tab=t.0#heading=h.4h01j4b8rjzv).
Overall, two new `.proto` files are introduced:
- `common.proto` - Shared types and messages between the worker specification & the new UDF protocol (to be introduced)
- `worker_spec.proto` - UDF worker specification
The changes introduced in this PR depend on [pull request 55089](https://github.com/apache/spark/pull/55089). As soon as [PR 55089](https://github.com/apache/spark/pull/55089) has been merged, this PR will be re-based on top.
### Why are the changes needed?
This is the first step toward a language-agnostic UDF protocol for Spark that enables UDF workers written in any language to communicate with the Spark engine through a well-defined specification and API boundary. The abstractions introduced here establish the core contract that concrete implementations (e.g., process-based or gRPC-based workers) will build on.
The worker specification introduced in this PR captures all the information Spark needs to:
- Plan UDF execution (concurrency, supported UDF types, etc.)
- Provision a UDF worker and connect to it for UDF invocation
### Does this PR introduce _any_ user-facing change?
No. All new APIs are marked @Experimental, and there are no behavioral changes to existing code.
### How was this patch tested?
- Compilation of the proto files verified via both Maven and SBT.
### Was this patch authored or co-authored using generative AI tooling?
Yes, in an assistive manner and for reviews.
--
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
Re: [PR] [SPARK-56284] Adding UDF worker specification protobuf definition [spark]
Posted by "hvanhovell (via GitHub)" <gi...@apache.org>.
hvanhovell commented on code in PR #55165:
URL: https://github.com/apache/spark/pull/55165#discussion_r3058790546
##########
udf/worker/proto/src/main/protobuf/worker_spec.proto:
##########
@@ -17,12 +17,194 @@
syntax = "proto3";
+import "common.proto";
+
package org.apache.spark.udf.worker;
option java_package = "org.apache.spark.udf.worker";
option java_multiple_files = true;
-// Placeholder -- full worker specification schema to be added.
-// See design doc: SPIP SPARK-55278.
+///
+/// Worker specification
+///
message UDFWorkerSpecification {
+ WorkerEnvironment environment = 1;
+ WorkerCapabilities capabilities = 2;
+
+ // How to create new workers.
+ // At the moment, only direct creation is supported.
+ // This can be extended with indirect/provisioned creation in the future.
+ oneof worker {
+ DirectWorker direct = 3;
+ }
+}
+
+message WorkerEnvironment {
+ // Callable that is responsible for environment setup.
+ optional ProcessCallable installation = 1;
+ // Callable, which is called to verify that an environment
+ // is suitable to start the UDF worker. This callable
+ // needs to verify that
+ // - The worker code itself is present
+ // - Any needed dependencies are present
+ optional ProcessCallable environment_verification = 2;
+ // Callable, which is invoked after the worker has been terminated.
+ // This can be used to cleanup dependencies or temporary resources.
+ // Be careful not to cleanup resources that could be used by
+ // other workers running in parallel.
+ optional ProcessCallable environment_cleanup = 3;
+}
+
+// Capabilities used for query planning
+message WorkerCapabilities {
+ // The data formats that the worker supports for UDF data in- & output.
+ // Every worker MUST at least support ARROW.
+ //
+ // It is expected that for each UDF execution, the input format
+ // always matches the output format.
+ //
+ // If a worker supports multiple data formats, the engine will select
+ // the most suitable one for each UDF invocation. Which format was chosen
+ // is reported by the engine as part of the UDF protocol's init message.
+ repeated UDFWorkerDataFormat supported_data_formats = 1;
+
+ // Which types of UDFs this worker supports.
+ // This should list all supported Shapes.
+ // The shape of a specific UDF will be communicated
+ // in the initial message of the UDF protocol.
+ //
+ // If an execution for an unsupported UDF type is requested
+ // the query will fail during query planning.
+ repeated UDFShape supported_udf_shapes = 2;
+
+ // Whether multiple, concurrent UDF
+ // connections are supported by this worker
+ // (for example via multi-threading).
+ //
+ // In the first implementation of the engine-side
+ // worker specification, this property will not be used.
+ //
+ // Usage of this property can be enabled in the future if the
+ // engine implements more advanced resource management (TBD).
+ //
+ bool supports_concurrent_udfs = 3;
+
+ // Whether compatible workers may be reused.
+ // If this is not supported, the worker is
+ // terminated after every single UDF invocation.
+ bool supports_reuse = 4;
+
+ // To be extended with UDF chaining, ...
+}
+
+// The worker that will be created to process UDFs
+message DirectWorker {
+ // Blocking callable that is terminated by SIGTERM/SIGKILL
+ ProcessCallable runner = 1;
+
+ UDFWorkerProperties properties = 2;
+}
+
+message UDFWorkerProperties {
+ // Maximum amount of time to wait until the worker can accept connections.
+ //
+ // The engine will use this timeout, if it does not exceed an
+ // engine-configurable maximum time (e.g. 30 seconds).
+ optional int32 initialization_timeout_ms = 1;
+
+ // Used for graceful engine-initiated termination signaled via SIGTERM.
+ // After this time, the worker process should have terminated itself.
+ // Otherwise, the process will be forcefully killed using SIGKILL.
+ //
+ // The engine will use this timeout, if it does not exceed an
+ // engine-configurable maximum time (e.g. 30 seconds).
+ optional int32 graceful_termination_timeout_ms = 2;
+
+ // The connection this [[DirectWorker]] supports. Note that a single
+ // connection is sufficient to run multiple UDFs and (gRPC) services.
+ //
+ // On [[DirectWorker]] creation, connection information
+ // is passed to the callable as a string parameter.
+ // The string format depends on the [[WorkerConnection]]:
+ //
+ // For example, when using TCP, the callable argument will be:
+ // --connection PORT
+ // Here is a concrete example
+ // --connection 8080
+ //
+ // For the format of each specific transport type, see the comments below.
+ WorkerConnection connection = 3;
+}
+
+message WorkerConnection {
+ oneof transport {
+ UnixDomainSocket unix_domain_socket = 1;
+ TcpConnection tcp = 2;
+ }
}
+
+// Communication between the engine and worker
+// is done using a UNIX domain socket.
+//
+// On [[DirectWorker]] creation, a path to a socket
+// to listen on is passed as an argument.
+// Examples:
+// /tmp/channel-uuid.sock
+// /some/system/path/channel-1234.sock
+message UnixDomainSocket {}
+
+// Communication between the engine and worker
+// is done using a localhost TCP connection.
+//
+// On [[DirectWorker]] creation, a PORT
+// is passed as a connection parameter.
+//
+// It is expected that the worker binds to this
+// port on both IPv4 and IPv6 localhost interfaces.
+// E.g. the worker server should be reachable via
+// 127.0.0.1:PORT and [::1]:PORT.
+//
+// Examples:
+// 8080
+// 1234
+message TcpConnection {}
Review Comment:
If this is local only, then name it as such?
--
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
Re: [PR] [SPARK-56284] Adding UDF worker specification protobuf definition [spark]
Posted by "sven-weber-db (via GitHub)" <gi...@apache.org>.
sven-weber-db commented on code in PR #55165:
URL: https://github.com/apache/spark/pull/55165#discussion_r3072639576
##########
udf/worker/proto/src/main/protobuf/worker_spec.proto:
##########
@@ -17,12 +17,194 @@
syntax = "proto3";
+import "common.proto";
+
package org.apache.spark.udf.worker;
option java_package = "org.apache.spark.udf.worker";
option java_multiple_files = true;
-// Placeholder -- full worker specification schema to be added.
-// See design doc: SPIP SPARK-55278.
+///
+/// Worker specification
+///
message UDFWorkerSpecification {
+ WorkerEnvironment environment = 1;
Review Comment:
The environment is not only needed for the `DirectWorker`. If we support a daemon-based worker creation in the future, this daemon would also need to be installed and set up in some way, which can be ensured through the environment. Therefore, I think this is not solely a property of the `DirectWorker`, but of all possible worker-creation modes we might support in the future.
--
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
Re: [PR] [SPARK-56284] Adding UDF worker specification protobuf definition [spark]
Posted by "hvanhovell (via GitHub)" <gi...@apache.org>.
hvanhovell commented on code in PR #55165:
URL: https://github.com/apache/spark/pull/55165#discussion_r3058833561
##########
udf/worker/proto/src/main/protobuf/worker_spec.proto:
##########
@@ -17,12 +17,194 @@
syntax = "proto3";
+import "common.proto";
+
package org.apache.spark.udf.worker;
option java_package = "org.apache.spark.udf.worker";
option java_multiple_files = true;
-// Placeholder -- full worker specification schema to be added.
-// See design doc: SPIP SPARK-55278.
+///
+/// Worker specification
+///
message UDFWorkerSpecification {
+ WorkerEnvironment environment = 1;
+ WorkerCapabilities capabilities = 2;
+
+ // How to create new workers.
+ // At the moment, only direct creation is supported.
+ // This can be extended with indirect/provisioned creation in the future.
+ oneof worker {
+ DirectWorker direct = 3;
+ }
+}
+
+message WorkerEnvironment {
Review Comment:
This needs documentation.
--
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
Re: [PR] [SPARK-56284] Adding UDF worker specification protobuf definition [spark]
Posted by "hvanhovell (via GitHub)" <gi...@apache.org>.
hvanhovell commented on code in PR #55165:
URL: https://github.com/apache/spark/pull/55165#discussion_r3073914155
##########
udf/worker/proto/src/main/protobuf/worker_spec.proto:
##########
@@ -17,12 +17,272 @@
syntax = "proto3";
+import "common.proto";
+
package org.apache.spark.udf.worker;
option java_package = "org.apache.spark.udf.worker";
option java_multiple_files = true;
-// Placeholder -- full worker specification schema to be added.
-// See design doc: SPIP SPARK-55278.
+///
+/// Worker specification
+///
message UDFWorkerSpecification {
+ // (Required)
+ WorkerEnvironment environment = 1;
+ // (Required)
+ WorkerCapabilities capabilities = 2;
+
+ // How to create new workers.
+ // At the moment, only direct creation is supported.
+ // This can be extended with indirect/provisioned creation in the future.
+ //
+ // (Required)
+ oneof worker {
+ DirectWorker direct = 3;
+ }
+}
+
+// Set of callables that can be used to setup, verify,
+// and cleanup the worker environment before the worker
+// callable is invoked.
+//
+// These callables should be specified if the worker
+// setup cannot be assumed to be pre-installed on the
+// target cluster.
+//
+// If defined, the below callables will be invoked once
+// per Spark executor in the following order (top -> bottom):
+//
+// ┌──────────────┐
+// │ │
+// │ Verification │
+// │ │
+// └──┬────────┬──┘
+// │ │
+// Failed │ │ Succeeded
+// │ │
+// ┌──────▼───────┐│
+// │ ││
+// │ Installation ││
+// │ ││
+// └──────┬───────┘│
+// │ │
+// ▼ ▼
+// ...
+// UDF worker creation
+// ...
+// │
+// │
+// ┌────▼────┐
+// │ │
+// │ Cleanup │
+// │ │
+// └─────────┘
+//
+// All scripts are optional.
+// However, if a verification script is supplied, an installation
Review Comment:
Unless the verification scripts checks for a fatal condition.
--
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
Re: [PR] [SPARK-56284] Adding UDF worker specification protobuf definition [spark]
Posted by "hvanhovell (via GitHub)" <gi...@apache.org>.
hvanhovell commented on code in PR #55165:
URL: https://github.com/apache/spark/pull/55165#discussion_r3058786589
##########
udf/worker/proto/src/main/protobuf/worker_spec.proto:
##########
@@ -17,12 +17,194 @@
syntax = "proto3";
+import "common.proto";
+
package org.apache.spark.udf.worker;
option java_package = "org.apache.spark.udf.worker";
option java_multiple_files = true;
-// Placeholder -- full worker specification schema to be added.
-// See design doc: SPIP SPARK-55278.
+///
+/// Worker specification
+///
message UDFWorkerSpecification {
+ WorkerEnvironment environment = 1;
+ WorkerCapabilities capabilities = 2;
+
+ // How to create new workers.
+ // At the moment, only direct creation is supported.
+ // This can be extended with indirect/provisioned creation in the future.
+ oneof worker {
+ DirectWorker direct = 3;
+ }
+}
+
+message WorkerEnvironment {
+ // Callable that is responsible for environment setup.
+ optional ProcessCallable installation = 1;
+ // Callable, which is called to verify that an environment
+ // is suitable to start the UDF worker. This callable
+ // needs to verify that
+ // - The worker code itself is present
+ // - Any needed dependencies are present
+ optional ProcessCallable environment_verification = 2;
+ // Callable, which is invoked after the worker has been terminated.
+ // This can be used to cleanup dependencies or temporary resources.
+ // Be careful not to cleanup resources that could be used by
+ // other workers running in parallel.
+ optional ProcessCallable environment_cleanup = 3;
+}
+
+// Capabilities used for query planning
+message WorkerCapabilities {
+ // The data formats that the worker supports for UDF data in- & output.
+ // Every worker MUST at least support ARROW.
+ //
+ // It is expected that for each UDF execution, the input format
+ // always matches the output format.
+ //
+ // If a worker supports multiple data formats, the engine will select
+ // the most suitable one for each UDF invocation. Which format was chosen
+ // is reported by the engine as part of the UDF protocol's init message.
+ repeated UDFWorkerDataFormat supported_data_formats = 1;
+
+ // Which types of UDFs this worker supports.
+ // This should list all supported Shapes.
+ // The shape of a specific UDF will be communicated
+ // in the initial message of the UDF protocol.
+ //
+ // If an execution for an unsupported UDF type is requested
+ // the query will fail during query planning.
+ repeated UDFShape supported_udf_shapes = 2;
+
+ // Whether multiple, concurrent UDF
+ // connections are supported by this worker
+ // (for example via multi-threading).
+ //
+ // In the first implementation of the engine-side
+ // worker specification, this property will not be used.
+ //
+ // Usage of this property can be enabled in the future if the
+ // engine implements more advanced resource management (TBD).
+ //
+ bool supports_concurrent_udfs = 3;
+
+ // Whether compatible workers may be reused.
+ // If this is not supported, the worker is
+ // terminated after every single UDF invocation.
+ bool supports_reuse = 4;
+
+ // To be extended with UDF chaining, ...
+}
+
+// The worker that will be created to process UDFs
+message DirectWorker {
+ // Blocking callable that is terminated by SIGTERM/SIGKILL
+ ProcessCallable runner = 1;
+
+ UDFWorkerProperties properties = 2;
+}
+
+message UDFWorkerProperties {
+ // Maximum amount of time to wait until the worker can accept connections.
+ //
+ // The engine will use this timeout, if it does not exceed an
+ // engine-configurable maximum time (e.g. 30 seconds).
+ optional int32 initialization_timeout_ms = 1;
+
+ // Used for graceful engine-initiated termination signaled via SIGTERM.
+ // After this time, the worker process should have terminated itself.
+ // Otherwise, the process will be forcefully killed using SIGKILL.
+ //
+ // The engine will use this timeout, if it does not exceed an
+ // engine-configurable maximum time (e.g. 30 seconds).
+ optional int32 graceful_termination_timeout_ms = 2;
+
+ // The connection this [[DirectWorker]] supports. Note that a single
+ // connection is sufficient to run multiple UDFs and (gRPC) services.
+ //
+ // On [[DirectWorker]] creation, connection information
+ // is passed to the callable as a string parameter.
+ // The string format depends on the [[WorkerConnection]]:
+ //
+ // For example, when using TCP, the callable argument will be:
+ // --connection PORT
+ // Here is a concrete example
+ // --connection 8080
+ //
+ // For the format of each specific transport type, see the comments below.
+ WorkerConnection connection = 3;
+}
+
+message WorkerConnection {
+ oneof transport {
+ UnixDomainSocket unix_domain_socket = 1;
+ TcpConnection tcp = 2;
+ }
}
+
+// Communication between the engine and worker
+// is done using a UNIX domain socket.
+//
+// On [[DirectWorker]] creation, a path to a socket
+// to listen on is passed as an argument.
+// Examples:
+// /tmp/channel-uuid.sock
+// /some/system/path/channel-1234.sock
+message UnixDomainSocket {}
Review Comment:
I am guessing you are going to add a path to this proto at some point?
--
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
Re: [PR] [SPARK-56284] Adding UDF worker specification protobuf definition [spark]
Posted by "hvanhovell (via GitHub)" <gi...@apache.org>.
hvanhovell commented on code in PR #55165:
URL: https://github.com/apache/spark/pull/55165#discussion_r3058760606
##########
udf/worker/proto/src/main/protobuf/common.proto:
##########
@@ -0,0 +1,51 @@
+/*
+ * 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 org.apache.spark.udf.worker;
+
+option java_package = "org.apache.spark.udf.worker";
+
+option java_multiple_files = true;
+
+// The UDF in & output data format.
+enum UDFWorkerDataFormat {
+ UDF_WORKER_DATA_FORMAT_UNSPECIFIED = 0;
+
+ // The worker accepts and produces Apache arrow batches.
+ ARROW = 1;
+}
+
+// The UDF execution type/shape.
+message UDFShape {
+ oneof shape {
Review Comment:
Why this indirection? The shape also has nothing to do with spark...
--
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
Re: [PR] [SPARK-56284] Adding UDF worker specification protobuf definition [spark]
Posted by "asf-gitbox-commits (via GitHub)" <gi...@apache.org>.
asf-gitbox-commits closed pull request #55165: [SPARK-56284] Adding UDF worker specification protobuf definition
URL: https://github.com/apache/spark/pull/55165
--
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
Re: [PR] [SPARK-56284] Adding UDF worker specification protobuf definition [spark]
Posted by "sven-weber-db (via GitHub)" <gi...@apache.org>.
sven-weber-db commented on code in PR #55165:
URL: https://github.com/apache/spark/pull/55165#discussion_r3072611663
##########
udf/worker/proto/src/main/protobuf/worker_spec.proto:
##########
@@ -17,12 +17,194 @@
syntax = "proto3";
+import "common.proto";
+
package org.apache.spark.udf.worker;
option java_package = "org.apache.spark.udf.worker";
option java_multiple_files = true;
-// Placeholder -- full worker specification schema to be added.
-// See design doc: SPIP SPARK-55278.
+///
+/// Worker specification
+///
message UDFWorkerSpecification {
+ WorkerEnvironment environment = 1;
+ WorkerCapabilities capabilities = 2;
+
+ // How to create new workers.
+ // At the moment, only direct creation is supported.
+ // This can be extended with indirect/provisioned creation in the future.
+ oneof worker {
+ DirectWorker direct = 3;
+ }
+}
+
+message WorkerEnvironment {
+ // Callable that is responsible for environment setup.
+ optional ProcessCallable installation = 1;
+ // Callable, which is called to verify that an environment
+ // is suitable to start the UDF worker. This callable
+ // needs to verify that
+ // - The worker code itself is present
+ // - Any needed dependencies are present
+ optional ProcessCallable environment_verification = 2;
+ // Callable, which is invoked after the worker has been terminated.
+ // This can be used to cleanup dependencies or temporary resources.
+ // Be careful not to cleanup resources that could be used by
+ // other workers running in parallel.
+ optional ProcessCallable environment_cleanup = 3;
+}
+
+// Capabilities used for query planning
+message WorkerCapabilities {
+ // The data formats that the worker supports for UDF data in- & output.
+ // Every worker MUST at least support ARROW.
+ //
+ // It is expected that for each UDF execution, the input format
+ // always matches the output format.
+ //
+ // If a worker supports multiple data formats, the engine will select
+ // the most suitable one for each UDF invocation. Which format was chosen
+ // is reported by the engine as part of the UDF protocol's init message.
+ repeated UDFWorkerDataFormat supported_data_formats = 1;
+
+ // Which types of UDFs this worker supports.
+ // This should list all supported Shapes.
+ // The shape of a specific UDF will be communicated
+ // in the initial message of the UDF protocol.
+ //
+ // If an execution for an unsupported UDF type is requested
+ // the query will fail during query planning.
+ repeated UDFShape supported_udf_shapes = 2;
+
+ // Whether multiple, concurrent UDF
+ // connections are supported by this worker
+ // (for example via multi-threading).
+ //
+ // In the first implementation of the engine-side
+ // worker specification, this property will not be used.
+ //
+ // Usage of this property can be enabled in the future if the
+ // engine implements more advanced resource management (TBD).
+ //
+ bool supports_concurrent_udfs = 3;
+
+ // Whether compatible workers may be reused.
+ // If this is not supported, the worker is
+ // terminated after every single UDF invocation.
+ bool supports_reuse = 4;
+
+ // To be extended with UDF chaining, ...
+}
+
+// The worker that will be created to process UDFs
+message DirectWorker {
+ // Blocking callable that is terminated by SIGTERM/SIGKILL
+ ProcessCallable runner = 1;
+
+ UDFWorkerProperties properties = 2;
+}
+
+message UDFWorkerProperties {
+ // Maximum amount of time to wait until the worker can accept connections.
+ //
+ // The engine will use this timeout, if it does not exceed an
+ // engine-configurable maximum time (e.g. 30 seconds).
+ optional int32 initialization_timeout_ms = 1;
+
+ // Used for graceful engine-initiated termination signaled via SIGTERM.
+ // After this time, the worker process should have terminated itself.
+ // Otherwise, the process will be forcefully killed using SIGKILL.
+ //
+ // The engine will use this timeout, if it does not exceed an
+ // engine-configurable maximum time (e.g. 30 seconds).
+ optional int32 graceful_termination_timeout_ms = 2;
+
+ // The connection this [[DirectWorker]] supports. Note that a single
+ // connection is sufficient to run multiple UDFs and (gRPC) services.
+ //
+ // On [[DirectWorker]] creation, connection information
+ // is passed to the callable as a string parameter.
+ // The string format depends on the [[WorkerConnection]]:
+ //
+ // For example, when using TCP, the callable argument will be:
+ // --connection PORT
+ // Here is a concrete example
+ // --connection 8080
+ //
+ // For the format of each specific transport type, see the comments below.
+ WorkerConnection connection = 3;
+}
+
+message WorkerConnection {
+ oneof transport {
+ UnixDomainSocket unix_domain_socket = 1;
+ TcpConnection tcp = 2;
+ }
}
+
+// Communication between the engine and worker
+// is done using a UNIX domain socket.
+//
+// On [[DirectWorker]] creation, a path to a socket
+// to listen on is passed as an argument.
+// Examples:
+// /tmp/channel-uuid.sock
+// /some/system/path/channel-1234.sock
+message UnixDomainSocket {}
Review Comment:
No, the path is injected into the worker callable via the `--connection` parameter. All properties in this `.proto` are client-supplied, but the path should remain in control of the engine.
I added a separate message for the UDS for future extension points. For UDS itself, it is unlikely that we will ever need an extension. However, for TCP, we might need to specify additional properties in the future. To make the proto consistent, I propose using the same structure for all transport types.
--
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
Re: [PR] [SPARK-56284] Adding UDF worker specification protobuf definition [spark]
Posted by "Yicong-Huang (via GitHub)" <gi...@apache.org>.
Yicong-Huang commented on code in PR #55165:
URL: https://github.com/apache/spark/pull/55165#discussion_r3075107667
##########
udf/worker/proto/src/main/protobuf/worker_spec.proto:
##########
@@ -17,12 +17,272 @@
syntax = "proto3";
+import "common.proto";
+
package org.apache.spark.udf.worker;
option java_package = "org.apache.spark.udf.worker";
option java_multiple_files = true;
-// Placeholder -- full worker specification schema to be added.
-// See design doc: SPIP SPARK-55278.
+///
+/// Worker specification
+///
message UDFWorkerSpecification {
+ // (Required)
+ WorkerEnvironment environment = 1;
+ // (Required)
+ WorkerCapabilities capabilities = 2;
+
+ // How to create new workers.
+ // At the moment, only direct creation is supported.
+ // This can be extended with indirect/provisioned creation in the future.
+ //
+ // (Required)
+ oneof worker {
+ DirectWorker direct = 3;
+ }
+}
+
+// Set of callables that can be used to setup, verify,
+// and cleanup the worker environment before the worker
+// callable is invoked.
+//
+// These callables should be specified if the worker
+// setup cannot be assumed to be pre-installed on the
+// target cluster.
+//
+// If defined, the below callables will be invoked once
+// per Spark executor in the following order (top -> bottom):
+//
+// ┌──────────────┐
+// │ │
+// │ Verification │
+// │ │
+// └──┬────────┬──┘
+// │ │
+// Failed │ │ Succeeded
+// │ │
+// ┌──────▼───────┐│
+// │ ││
+// │ Installation ││
+// │ ││
+// └──────┬───────┘│
+// │ │
+// ▼ ▼
+// ...
+// UDF worker creation
+// ...
+// │
+// │
+// ┌────▼────┐
+// │ │
+// │ Cleanup │
+// │ │
+// └─────────┘
+//
+// All scripts are optional.
+// However, if a verification script is supplied, an installation
+// script needs to be supplied as well.
+//
+message WorkerEnvironment {
+ // A callable that prepares cluster nodes by installing
+ // the required runtime, dependencies, and UDF worker binaries
+ // at the expected locations. This step ensures that
+ // the worker starting callable
+ // can subsequently locate and launch the worker process.
+ //
+ // (Required if environment_verification is given)
+ optional ProcessCallable installation = 1;
+
+ // Callable, which is called to verify that an environment
+ // is suitable to start the UDF worker. This callable
+ // needs to verify that
+ // - The worker code itself is present
+ // - Any needed dependencies are present
+ //
+ // (Optional)
+ optional ProcessCallable environment_verification = 2;
+
+ // Callable, which is invoked after all workers have been terminated.
+ // This can be used to cleanup dependencies or temporary resources.
+ //
+ // (Optional)
+ optional ProcessCallable environment_cleanup = 3;
Review Comment:
when installation fails, do we also clean up partial setup environment?
##########
udf/worker/proto/src/main/protobuf/worker_spec.proto:
##########
@@ -17,12 +17,272 @@
syntax = "proto3";
+import "common.proto";
+
package org.apache.spark.udf.worker;
option java_package = "org.apache.spark.udf.worker";
option java_multiple_files = true;
-// Placeholder -- full worker specification schema to be added.
-// See design doc: SPIP SPARK-55278.
+///
+/// Worker specification
+///
message UDFWorkerSpecification {
+ // (Required)
+ WorkerEnvironment environment = 1;
+ // (Required)
+ WorkerCapabilities capabilities = 2;
+
+ // How to create new workers.
+ // At the moment, only direct creation is supported.
+ // This can be extended with indirect/provisioned creation in the future.
+ //
+ // (Required)
+ oneof worker {
+ DirectWorker direct = 3;
+ }
+}
+
+// Set of callables that can be used to setup, verify,
+// and cleanup the worker environment before the worker
+// callable is invoked.
+//
+// These callables should be specified if the worker
+// setup cannot be assumed to be pre-installed on the
+// target cluster.
+//
+// If defined, the below callables will be invoked once
+// per Spark executor in the following order (top -> bottom):
+//
+// ┌──────────────┐
+// │ │
+// │ Verification │
+// │ │
+// └──┬────────┬──┘
+// │ │
+// Failed │ │ Succeeded
+// │ │
+// ┌──────▼───────┐│
+// │ ││
+// │ Installation ││
+// │ ││
+// └──────┬───────┘│
+// │ │
+// ▼ ▼
+// ...
+// UDF worker creation
+// ...
+// │
+// │
+// ┌────▼────┐
+// │ │
+// │ Cleanup │
+// │ │
+// └─────────┘
+//
+// All scripts are optional.
+// However, if a verification script is supplied, an installation
+// script needs to be supplied as well.
+//
+message WorkerEnvironment {
+ // A callable that prepares cluster nodes by installing
+ // the required runtime, dependencies, and UDF worker binaries
+ // at the expected locations. This step ensures that
+ // the worker starting callable
+ // can subsequently locate and launch the worker process.
+ //
+ // (Required if environment_verification is given)
+ optional ProcessCallable installation = 1;
+
+ // Callable, which is called to verify that an environment
+ // is suitable to start the UDF worker. This callable
+ // needs to verify that
+ // - The worker code itself is present
+ // - Any needed dependencies are present
+ //
+ // (Optional)
+ optional ProcessCallable environment_verification = 2;
+
+ // Callable, which is invoked after all workers have been terminated.
+ // This can be used to cleanup dependencies or temporary resources.
+ //
+ // (Optional)
+ optional ProcessCallable environment_cleanup = 3;
+}
+
+// Capabilities used for query planning
+// and running the worker during query execution.
+message WorkerCapabilities {
+ // The data formats that the worker supports for UDF data in- & output.
+ // Every worker MUST at least support ARROW.
+ //
+ // It is expected that for each UDF execution, the input format
+ // always matches the output format.
+ //
+ // If a worker supports multiple data formats, the engine will select
+ // the most suitable one for each UDF invocation. Which format was chosen
+ // is reported by the engine as part of the UDF protocol's init message.
+ //
+ // (Required)
+ repeated UDFWorkerDataFormat supported_data_formats = 1;
+
+ // Which UDF protocol communication patterns the worker
+ // supports. This should list all supported patterns.
+ // The pattern used for a specific UDF will be communicated
+ // in the initial message of the UDF protocol.
+ //
+ // If an execution for an unsupported pattern is requested
+ // the query will fail during query planning.
+ //
+ // (Required)
+ repeated UDFProtoCommunicationPattern supported_communication_patterns = 2;
+
+ // Whether multiple, concurrent UDF
+ // connections are supported by this worker
+ // (for example via multi-threading).
+ //
+ // In the first implementation of the engine-side
+ // worker specification, this property will not be used.
+ //
+ // Usage of this property can be enabled in the future if the
+ // engine implements more advanced resource management (TBD).
+ //
+ // (Optional)
+ optional bool supports_concurrent_udfs = 3;
+
+ // Whether compatible workers may be reused.
+ // If this is not supported, the worker is
+ // terminated after every single UDF invocation.
+ //
+ // (Optional)
+ optional bool supports_reuse = 4;
+
+ // To be extended with UDF chaining, ...
+}
+
+// The worker that will be created to process UDFs
+message DirectWorker {
+ // Blocking callable that is terminated by SIGTERM/SIGKILL
+ //
+ // (Required)
+ ProcessCallable runner = 1;
+
+ // (Required)
+ UDFWorkerProperties properties = 2;
}
+
+message UDFWorkerProperties {
+ // Maximum amount of time to wait until the worker can accept connections.
+ //
+ // The engine will use this timeout, if it does not exceed an
+ // engine-configurable maximum time (e.g. 30 seconds).
+ //
+ // (Optional)
+ optional int32 initialization_timeout_ms = 1;
Review Comment:
or maybe protobuf.Duration
##########
udf/worker/proto/src/main/protobuf/worker_spec.proto:
##########
@@ -17,12 +17,272 @@
syntax = "proto3";
+import "common.proto";
+
package org.apache.spark.udf.worker;
option java_package = "org.apache.spark.udf.worker";
option java_multiple_files = true;
-// Placeholder -- full worker specification schema to be added.
-// See design doc: SPIP SPARK-55278.
+///
+/// Worker specification
+///
message UDFWorkerSpecification {
+ // (Required)
+ WorkerEnvironment environment = 1;
+ // (Required)
+ WorkerCapabilities capabilities = 2;
+
+ // How to create new workers.
+ // At the moment, only direct creation is supported.
+ // This can be extended with indirect/provisioned creation in the future.
+ //
+ // (Required)
+ oneof worker {
+ DirectWorker direct = 3;
+ }
+}
+
+// Set of callables that can be used to setup, verify,
+// and cleanup the worker environment before the worker
+// callable is invoked.
+//
+// These callables should be specified if the worker
+// setup cannot be assumed to be pre-installed on the
+// target cluster.
+//
+// If defined, the below callables will be invoked once
+// per Spark executor in the following order (top -> bottom):
+//
+// ┌──────────────┐
+// │ │
+// │ Verification │
+// │ │
+// └──┬────────┬──┘
+// │ │
+// Failed │ │ Succeeded
+// │ │
+// ┌──────▼───────┐│
+// │ ││
+// │ Installation ││
+// │ ││
+// └──────┬───────┘│
+// │ │
+// ▼ ▼
+// ...
+// UDF worker creation
+// ...
+// │
+// │
+// ┌────▼────┐
+// │ │
+// │ Cleanup │
+// │ │
+// └─────────┘
+//
+// All scripts are optional.
+// However, if a verification script is supplied, an installation
+// script needs to be supplied as well.
+//
+message WorkerEnvironment {
+ // A callable that prepares cluster nodes by installing
+ // the required runtime, dependencies, and UDF worker binaries
+ // at the expected locations. This step ensures that
+ // the worker starting callable
+ // can subsequently locate and launch the worker process.
+ //
+ // (Required if environment_verification is given)
+ optional ProcessCallable installation = 1;
+
+ // Callable, which is called to verify that an environment
+ // is suitable to start the UDF worker. This callable
+ // needs to verify that
+ // - The worker code itself is present
+ // - Any needed dependencies are present
+ //
+ // (Optional)
+ optional ProcessCallable environment_verification = 2;
+
+ // Callable, which is invoked after all workers have been terminated.
+ // This can be used to cleanup dependencies or temporary resources.
+ //
+ // (Optional)
+ optional ProcessCallable environment_cleanup = 3;
+}
+
+// Capabilities used for query planning
+// and running the worker during query execution.
+message WorkerCapabilities {
+ // The data formats that the worker supports for UDF data in- & output.
+ // Every worker MUST at least support ARROW.
+ //
+ // It is expected that for each UDF execution, the input format
+ // always matches the output format.
+ //
+ // If a worker supports multiple data formats, the engine will select
+ // the most suitable one for each UDF invocation. Which format was chosen
+ // is reported by the engine as part of the UDF protocol's init message.
+ //
+ // (Required)
+ repeated UDFWorkerDataFormat supported_data_formats = 1;
+
+ // Which UDF protocol communication patterns the worker
+ // supports. This should list all supported patterns.
+ // The pattern used for a specific UDF will be communicated
+ // in the initial message of the UDF protocol.
+ //
+ // If an execution for an unsupported pattern is requested
+ // the query will fail during query planning.
+ //
+ // (Required)
+ repeated UDFProtoCommunicationPattern supported_communication_patterns = 2;
+
+ // Whether multiple, concurrent UDF
+ // connections are supported by this worker
+ // (for example via multi-threading).
+ //
+ // In the first implementation of the engine-side
+ // worker specification, this property will not be used.
+ //
+ // Usage of this property can be enabled in the future if the
+ // engine implements more advanced resource management (TBD).
+ //
+ // (Optional)
+ optional bool supports_concurrent_udfs = 3;
Review Comment:
if not used right now, we can just reserve the keyword https://protobuf.dev/programming-guides/proto3/ ?
e.g.,
```
message Foo {
reserved 2, 15, 9 to 11;
}
```
##########
udf/worker/proto/src/main/protobuf/worker_spec.proto:
##########
@@ -17,12 +17,272 @@
syntax = "proto3";
+import "common.proto";
+
package org.apache.spark.udf.worker;
option java_package = "org.apache.spark.udf.worker";
option java_multiple_files = true;
-// Placeholder -- full worker specification schema to be added.
-// See design doc: SPIP SPARK-55278.
+///
+/// Worker specification
+///
message UDFWorkerSpecification {
+ // (Required)
+ WorkerEnvironment environment = 1;
+ // (Required)
+ WorkerCapabilities capabilities = 2;
+
+ // How to create new workers.
+ // At the moment, only direct creation is supported.
+ // This can be extended with indirect/provisioned creation in the future.
+ //
+ // (Required)
+ oneof worker {
+ DirectWorker direct = 3;
+ }
+}
+
+// Set of callables that can be used to setup, verify,
+// and cleanup the worker environment before the worker
+// callable is invoked.
+//
+// These callables should be specified if the worker
+// setup cannot be assumed to be pre-installed on the
+// target cluster.
+//
+// If defined, the below callables will be invoked once
+// per Spark executor in the following order (top -> bottom):
+//
+// ┌──────────────┐
+// │ │
+// │ Verification │
+// │ │
+// └──┬────────┬──┘
+// │ │
+// Failed │ │ Succeeded
+// │ │
+// ┌──────▼───────┐│
+// │ ││
+// │ Installation ││
+// │ ││
+// └──────┬───────┘│
+// │ │
+// ▼ ▼
+// ...
+// UDF worker creation
+// ...
Review Comment:
it is not clear how installation failures are handled. also (not related to this figure) it is not clear how worker creation failures are handled.
##########
udf/worker/proto/src/main/protobuf/worker_spec.proto:
##########
@@ -17,12 +17,272 @@
syntax = "proto3";
+import "common.proto";
+
package org.apache.spark.udf.worker;
option java_package = "org.apache.spark.udf.worker";
option java_multiple_files = true;
-// Placeholder -- full worker specification schema to be added.
-// See design doc: SPIP SPARK-55278.
+///
+/// Worker specification
+///
message UDFWorkerSpecification {
+ // (Required)
+ WorkerEnvironment environment = 1;
+ // (Required)
+ WorkerCapabilities capabilities = 2;
+
+ // How to create new workers.
+ // At the moment, only direct creation is supported.
+ // This can be extended with indirect/provisioned creation in the future.
+ //
+ // (Required)
+ oneof worker {
+ DirectWorker direct = 3;
+ }
+}
+
+// Set of callables that can be used to setup, verify,
+// and cleanup the worker environment before the worker
+// callable is invoked.
+//
+// These callables should be specified if the worker
+// setup cannot be assumed to be pre-installed on the
+// target cluster.
+//
+// If defined, the below callables will be invoked once
+// per Spark executor in the following order (top -> bottom):
+//
+// ┌──────────────┐
+// │ │
+// │ Verification │
+// │ │
+// └──┬────────┬──┘
+// │ │
+// Failed │ │ Succeeded
+// │ │
+// ┌──────▼───────┐│
+// │ ││
+// │ Installation ││
+// │ ││
+// └──────┬───────┘│
+// │ │
+// ▼ ▼
+// ...
+// UDF worker creation
+// ...
+// │
+// │
+// ┌────▼────┐
+// │ │
+// │ Cleanup │
+// │ │
+// └─────────┘
+//
+// All scripts are optional.
+// However, if a verification script is supplied, an installation
+// script needs to be supplied as well.
+//
+message WorkerEnvironment {
+ // A callable that prepares cluster nodes by installing
+ // the required runtime, dependencies, and UDF worker binaries
+ // at the expected locations. This step ensures that
+ // the worker starting callable
+ // can subsequently locate and launch the worker process.
+ //
+ // (Required if environment_verification is given)
+ optional ProcessCallable installation = 1;
+
+ // Callable, which is called to verify that an environment
+ // is suitable to start the UDF worker. This callable
+ // needs to verify that
+ // - The worker code itself is present
+ // - Any needed dependencies are present
+ //
+ // (Optional)
+ optional ProcessCallable environment_verification = 2;
+
+ // Callable, which is invoked after all workers have been terminated.
+ // This can be used to cleanup dependencies or temporary resources.
+ //
+ // (Optional)
+ optional ProcessCallable environment_cleanup = 3;
+}
+
+// Capabilities used for query planning
+// and running the worker during query execution.
+message WorkerCapabilities {
+ // The data formats that the worker supports for UDF data in- & output.
+ // Every worker MUST at least support ARROW.
+ //
+ // It is expected that for each UDF execution, the input format
+ // always matches the output format.
+ //
+ // If a worker supports multiple data formats, the engine will select
+ // the most suitable one for each UDF invocation. Which format was chosen
+ // is reported by the engine as part of the UDF protocol's init message.
+ //
+ // (Required)
+ repeated UDFWorkerDataFormat supported_data_formats = 1;
+
+ // Which UDF protocol communication patterns the worker
+ // supports. This should list all supported patterns.
+ // The pattern used for a specific UDF will be communicated
+ // in the initial message of the UDF protocol.
+ //
+ // If an execution for an unsupported pattern is requested
+ // the query will fail during query planning.
+ //
+ // (Required)
+ repeated UDFProtoCommunicationPattern supported_communication_patterns = 2;
+
+ // Whether multiple, concurrent UDF
+ // connections are supported by this worker
+ // (for example via multi-threading).
+ //
+ // In the first implementation of the engine-side
+ // worker specification, this property will not be used.
+ //
+ // Usage of this property can be enabled in the future if the
+ // engine implements more advanced resource management (TBD).
+ //
+ // (Optional)
+ optional bool supports_concurrent_udfs = 3;
+
+ // Whether compatible workers may be reused.
+ // If this is not supported, the worker is
+ // terminated after every single UDF invocation.
+ //
+ // (Optional)
+ optional bool supports_reuse = 4;
+
+ // To be extended with UDF chaining, ...
+}
+
+// The worker that will be created to process UDFs
+message DirectWorker {
+ // Blocking callable that is terminated by SIGTERM/SIGKILL
+ //
+ // (Required)
+ ProcessCallable runner = 1;
+
+ // (Required)
+ UDFWorkerProperties properties = 2;
}
+
+message UDFWorkerProperties {
+ // Maximum amount of time to wait until the worker can accept connections.
+ //
+ // The engine will use this timeout, if it does not exceed an
+ // engine-configurable maximum time (e.g. 30 seconds).
+ //
+ // (Optional)
+ optional int32 initialization_timeout_ms = 1;
Review Comment:
timeout better use uint to avoid negative value?
--
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
Re: [PR] [SPARK-56284] Adding UDF worker specification protobuf definition [spark]
Posted by "sven-weber-db (via GitHub)" <gi...@apache.org>.
sven-weber-db commented on code in PR #55165:
URL: https://github.com/apache/spark/pull/55165#discussion_r3072573086
##########
udf/worker/proto/src/main/protobuf/common.proto:
##########
@@ -0,0 +1,51 @@
+/*
+ * 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 org.apache.spark.udf.worker;
+
+option java_package = "org.apache.spark.udf.worker";
+
+option java_multiple_files = true;
+
+// The UDF in & output data format.
+enum UDFWorkerDataFormat {
+ UDF_WORKER_DATA_FORMAT_UNSPECIFIED = 0;
+
+ // The worker accepts and produces Apache arrow batches.
+ ARROW = 1;
+}
+
+// The UDF execution type/shape.
+message UDFShape {
+ oneof shape {
Review Comment:
I revisited this Part with @haiyangsun-db. Instead of using a `UDFShape`, we now propose a `UDFProtoCommunicationPattern`. This captures which communication pattern of the UDFProto (to be added) is supported by this UDF worker. For all initial versions, this will be a bidirectional stream of bytes. However, for simpler use cases, we might add new communication modes, such as Request->Response, in the future.
The `UDFShape` will be covered via the client-specific init message. E.g., for Python UDFs, the Python client can tell the Python worker how to invoke the UDF (batch iterator, row-wise, etc.).
--
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
Re: [PR] [SPARK-56284] Adding UDF worker specification protobuf definition [spark]
Posted by "sven-weber-db (via GitHub)" <gi...@apache.org>.
sven-weber-db commented on code in PR #55165:
URL: https://github.com/apache/spark/pull/55165#discussion_r3045940264
##########
udf/worker/proto/src/main/protobuf/worker_spec.proto:
##########
@@ -17,12 +17,186 @@
syntax = "proto3";
+import "common.proto";
+
package org.apache.spark.udf.worker;
option java_package = "org.apache.spark.udf.worker";
option java_multiple_files = true;
-// Placeholder -- full worker specification schema to be added.
-// See design doc: SPIP SPARK-55278.
+///
+/// Worker specification
+///
message UDFWorkerSpecification {
+ WorkerEnvironment environment = 1;
+ WorkerCapabilities capabilities = 2;
+
+ // How to create new workers.
+ // At the moment, only direct creation is supported.
+ // This can be extended with indirect/provisioned creation in the future.
+ oneof worker {
+ DirectWorker direct = 3;
+ }
+}
+
+message WorkerEnvironment {
+ // Callable that is responsible for environment setup.
+ optional ProcessCallable installation = 1;
+ // Callable, which is called to verify that an environment
+ // is suitable to start the UDF worker. This callable
+ // needs to verify that
+ // - The worker code itself is present
+ // - Any needed dependencies are present
+ optional ProcessCallable environment_verification = 2;
+ // Callable, which is invoked after the worker has been terminated.
+ // This can be used to cleanup dependencies or temporary resources.
+ // Be careful not to cleanup resources that could be used by
+ // other workers running in parallel.
+ optional ProcessCallable environment_cleanup = 3;
+}
+
+// Capabilities used for query planning
+message WorkerCapabilities {
+ // The data formats that the worker supports for UDF data in- & output.
+ // Every worker MUST at least support ARROW.
+ //
+ // It is expected that for each UDF execution, the input format
+ // always matches the output format.
+ //
+ // If a worker supports multiple data formats, the engine will select
+ // the most suitable one for each UDF invocation. Which format was chosen
+ // is reported by the engine as part of the UDF protocol's init message.
+ repeated UDFWorkerDataFormat supported_data_formats = 1;
+
+ // Which types of UDFs this worker supports.
+ // This should list all supported Shapes.
+ // Of which shape a specific UDF is will be communicated
+ // in the initial message of the UDF protocol.
+ //
+ // If a execution for an unsupported UDF type is requested
+ // the query will fail during query planning.
+ repeated UDFShape supported_udf_shapes = 2;
+
+ // Whether multiple, concurrent UDF
+ // connections are supported by this worker
+ // (for example via multi-threading).
+ //
+ // In the first implementation of the engine-side
+ // worker specification, this property will not be used.
+ //
+ // Usage of this property, can be enabled in the future if the
+ // engine implements more advanced resource management (Tbd).
+ //
+ bool supports_concurrent_udfs = 3;
+
+ // Whether compatible workers may be reused.
+ // If this is not supported, the worker is
+ // terminated after every single UDF invocation.
+ bool supports_reuse = 4;
+
+ // To be extended with UDF chaining, ...
+}
+
+// The worker that will be created to process UDFs
+message DirectWorker {
+ // Blocking callable that is terminated by SIGTERM/SIGKILL
+ ProcessCallable runner = 1;
+
+ UDFWorkerProperties properties = 2;
+}
+
+message UDFWorkerProperties {
+ // Maximum amount of time to wait until the worker can accept connections.
+ //
+ // The engine will use this timeout, if it does not exceed a
+ // engine-configurable maximum time (e.g. 30 seconds).
+ optional int32 initialization_timeout_ms = 1;
+
+ // Used for graceful engine-initiated termination signaled via SIGTERM.
+ // After this time, the worker process should have terminated itself.
+ // Otherwise, the process will be forcefully killed using SIGKILL.
+ //
+ // The engine will use this timeout, if it does not exceed a
+ // engine-configurable maximum time (e.g. 30 seconds).
+ optional int32 graceful_termination_timeout_ms = 2;
+
+ // The connection this [[UDFWorker]] supports. Note that a single
Review Comment:
Good catch! `[[UDFWorker]]` was the name we previously used for `DirectWorker`. Before raising this PR, it was renamed, and I seem to have forgotten to update all references in the text to the old name. This should be fixed now. Thank you!
##########
udf/worker/proto/src/main/protobuf/common.proto:
##########
@@ -0,0 +1,51 @@
+/*
+ * 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 org.apache.spark.udf.worker;
+
+option java_package = "org.apache.spark.udf.worker";
+
+option java_multiple_files = true;
+
+// The UDF in & output data format.
+enum UDFWorkerDataFormat {
+ UDF_WORKER_DATA_FORMAT_UNSPECIFIED = 0;
+
+ // The worker accepts and produces Apache arrow batches.
+ ARROW = 1;
+}
+
+// The UDF execution type/shape.
+message UDFShape {
+ oneof shape {
+ SparkUDFShapes spark = 1;
+ }
+}
+
+enum SparkUDFShapes {
Review Comment:
Good catch, thank you! Fixed.
##########
udf/worker/proto/src/main/protobuf/common.proto:
##########
@@ -0,0 +1,51 @@
+/*
+ * 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 org.apache.spark.udf.worker;
+
+option java_package = "org.apache.spark.udf.worker";
+
+option java_multiple_files = true;
+
+// The UDF in & output data format.
+enum UDFWorkerDataFormat {
+ UDF_WORKER_DATA_FORMAT_UNSPECIFIED = 0;
+
+ // The worker accepts and produces Apache arrow batches.
+ ARROW = 1;
+}
+
+// The UDF execution type/shape.
+message UDFShape {
+ oneof shape {
+ SparkUDFShapes spark = 1;
+ }
+}
+
+enum SparkUDFShapes {
+ SPARK_UDF_SHAPE_UNSPECIFIED = 0;
+
+ // UDF receives a row with 0+ columns as input
+ // and produces a single, scalar value as output
+ EXPRESSION = 1;
+
+ // UDF receives a iterator to batch of rows as input and
+ // produces iterator to a batch of rows as output.
+ MAP_PARTITIONS = 2;
Review Comment:
Missed this - thank you!
##########
udf/worker/proto/src/main/protobuf/worker_spec.proto:
##########
@@ -17,12 +17,186 @@
syntax = "proto3";
+import "common.proto";
+
package org.apache.spark.udf.worker;
option java_package = "org.apache.spark.udf.worker";
option java_multiple_files = true;
-// Placeholder -- full worker specification schema to be added.
-// See design doc: SPIP SPARK-55278.
+///
+/// Worker specification
+///
message UDFWorkerSpecification {
+ WorkerEnvironment environment = 1;
+ WorkerCapabilities capabilities = 2;
+
+ // How to create new workers.
+ // At the moment, only direct creation is supported.
+ // This can be extended with indirect/provisioned creation in the future.
+ oneof worker {
+ DirectWorker direct = 3;
+ }
+}
+
+message WorkerEnvironment {
+ // Callable that is responsible for environment setup.
+ optional ProcessCallable installation = 1;
+ // Callable, which is called to verify that an environment
+ // is suitable to start the UDF worker. This callable
+ // needs to verify that
+ // - The worker code itself is present
+ // - Any needed dependencies are present
+ optional ProcessCallable environment_verification = 2;
+ // Callable, which is invoked after the worker has been terminated.
+ // This can be used to cleanup dependencies or temporary resources.
+ // Be careful not to cleanup resources that could be used by
+ // other workers running in parallel.
+ optional ProcessCallable environment_cleanup = 3;
+}
+
+// Capabilities used for query planning
+message WorkerCapabilities {
+ // The data formats that the worker supports for UDF data in- & output.
+ // Every worker MUST at least support ARROW.
+ //
+ // It is expected that for each UDF execution, the input format
+ // always matches the output format.
+ //
+ // If a worker supports multiple data formats, the engine will select
+ // the most suitable one for each UDF invocation. Which format was chosen
+ // is reported by the engine as part of the UDF protocol's init message.
+ repeated UDFWorkerDataFormat supported_data_formats = 1;
+
+ // Which types of UDFs this worker supports.
+ // This should list all supported Shapes.
+ // Of which shape a specific UDF is will be communicated
+ // in the initial message of the UDF protocol.
+ //
+ // If a execution for an unsupported UDF type is requested
+ // the query will fail during query planning.
+ repeated UDFShape supported_udf_shapes = 2;
+
+ // Whether multiple, concurrent UDF
+ // connections are supported by this worker
+ // (for example via multi-threading).
+ //
+ // In the first implementation of the engine-side
+ // worker specification, this property will not be used.
+ //
+ // Usage of this property, can be enabled in the future if the
+ // engine implements more advanced resource management (Tbd).
+ //
+ bool supports_concurrent_udfs = 3;
+
+ // Whether compatible workers may be reused.
+ // If this is not supported, the worker is
+ // terminated after every single UDF invocation.
+ bool supports_reuse = 4;
+
+ // To be extended with UDF chaining, ...
+}
+
+// The worker that will be created to process UDFs
+message DirectWorker {
+ // Blocking callable that is terminated by SIGTERM/SIGKILL
+ ProcessCallable runner = 1;
+
+ UDFWorkerProperties properties = 2;
+}
+
+message UDFWorkerProperties {
+ // Maximum amount of time to wait until the worker can accept connections.
+ //
+ // The engine will use this timeout, if it does not exceed a
+ // engine-configurable maximum time (e.g. 30 seconds).
+ optional int32 initialization_timeout_ms = 1;
+
+ // Used for graceful engine-initiated termination signaled via SIGTERM.
+ // After this time, the worker process should have terminated itself.
+ // Otherwise, the process will be forcefully killed using SIGKILL.
+ //
+ // The engine will use this timeout, if it does not exceed a
+ // engine-configurable maximum time (e.g. 30 seconds).
+ optional int32 graceful_termination_timeout_ms = 2;
+
+ // The connection this [[UDFWorker]] supports. Note that a single
+ // connection is sufficient to run multiple UDFs and (gRPC) services.
+ //
+ // On [[UDFWorker]] creation, connection information
+ // is passed to the callable as a string parameter.
+ // The string format depends on the [[WorkerConnection]]:
+ //
+ // For example, when using TCP, the callable argument will be:
+ // --connection PORT
+ // Here is a concrete example
+ // --connection 8080
+ //
+ // For the format of each specific transport type, see the comments below.
+ WorkerConnection connection = 3;
+}
+
+message WorkerConnection {
+ oneof transport {
+ UnixDomainSocket unix_domain_socket = 1;
+ TcpConnection tcp = 2;
+ }
}
+
+ // Communication between the engine and worker
+// is done using a UNIX domain socket.
+//
+// On [[UDFWorker]] creation, a path to a socket
+// to listen on is passed as a argument.
+// Examples:
+// /tmp/channel-uuid.sock
+// /some/system/path/channel-1234.sock
+message UnixDomainSocket {}
+
+// Communication between the engine and worker
+// is done using a localhost TCP connection.
+//
+// On [[UDFWorker]] creation, a PORT
+// is passed as a connection parameter.
+//
+// It is expected that the worker binds to this
+// port on both IPv4 and IPv6 localhost interfaces.
+// E.g. the worker server should be reachable via
+// 127.0.0.1:PORT and [::1]:PORT.
+//
+// Examples:
+// 8080
+// 1234
+message TcpConnection {}
+
+message ProcessCallable {
+ // Executable to invoke.
+ // Examples:
+ // ["python3", "-m"]
+ // ["worker.bin"]
+ // ["java", "worker.java"]
+ // ["bin/bash", "-c"]
+ // This executable should be blocking, until the task is finished.
+ // Termination is requested via a SIGTERM signal.
+ //
+ // Success/Failure can be indicated via exit codes:
+ // Exit code 0 -> Success
+ // Exit code != 0 -> Failure
+ repeated string command = 1;
+
+ // Arguments passed directly to the executable.
+ // Examples:
+ // ["udf_worker.py"]
+ // [""]
+ // ["--max_concurrency", "5"]
+ // ["\"echo 'Test'\""]
+ //
+ // Every executable will ALWAYS receive a
+ // --id argument. This argument CANNOT be part of the below list of arguments.
Review Comment:
Yes, very good point. I have updated the description to a list of restricted values including both ``--id`` and ``--connection``. Thank you!
--
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
Re: [PR] [SPARK-56284] Adding UDF worker specification protobuf definition [spark]
Posted by "hvanhovell (via GitHub)" <gi...@apache.org>.
hvanhovell commented on code in PR #55165:
URL: https://github.com/apache/spark/pull/55165#discussion_r3058764131
##########
udf/worker/proto/src/main/protobuf/common.proto:
##########
@@ -0,0 +1,51 @@
+/*
+ * 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 org.apache.spark.udf.worker;
+
+option java_package = "org.apache.spark.udf.worker";
+
+option java_multiple_files = true;
+
+// The UDF in & output data format.
+enum UDFWorkerDataFormat {
+ UDF_WORKER_DATA_FORMAT_UNSPECIFIED = 0;
+
+ // The worker accepts and produces Apache arrow batches.
+ ARROW = 1;
+}
+
+// The UDF execution type/shape.
+message UDFShape {
+ oneof shape {
+ SparkUDFShape spark = 1;
+ }
+}
+
+enum SparkUDFShape {
+ SPARK_UDF_SHAPE_UNSPECIFIED = 0;
+
+ // UDF receives a row with 0+ columns as input
+ // and produces a single, scalar value as output
+ EXPRESSION = 1;
Review Comment:
Name it `SCALAR` or `ONE_TO_ONE`?
--
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
Re: [PR] [SPARK-56284] Adding UDF worker specification protobuf definition [spark]
Posted by "sven-weber-db (via GitHub)" <gi...@apache.org>.
sven-weber-db commented on code in PR #55165:
URL: https://github.com/apache/spark/pull/55165#discussion_r3072574711
##########
udf/worker/proto/src/main/protobuf/common.proto:
##########
@@ -0,0 +1,51 @@
+/*
+ * 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 org.apache.spark.udf.worker;
+
+option java_package = "org.apache.spark.udf.worker";
+
+option java_multiple_files = true;
+
+// The UDF in & output data format.
+enum UDFWorkerDataFormat {
+ UDF_WORKER_DATA_FORMAT_UNSPECIFIED = 0;
+
+ // The worker accepts and produces Apache arrow batches.
+ ARROW = 1;
+}
+
+// The UDF execution type/shape.
+message UDFShape {
+ oneof shape {
+ SparkUDFShape spark = 1;
+ }
+}
+
+enum SparkUDFShape {
+ SPARK_UDF_SHAPE_UNSPECIFIED = 0;
+
+ // UDF receives a row with 0+ columns as input
+ // and produces a single, scalar value as output
+ EXPRESSION = 1;
Review Comment:
Deprecated according to the above comment
##########
udf/worker/proto/src/main/protobuf/common.proto:
##########
@@ -0,0 +1,51 @@
+/*
+ * 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 org.apache.spark.udf.worker;
+
+option java_package = "org.apache.spark.udf.worker";
+
+option java_multiple_files = true;
+
+// The UDF in & output data format.
+enum UDFWorkerDataFormat {
+ UDF_WORKER_DATA_FORMAT_UNSPECIFIED = 0;
+
+ // The worker accepts and produces Apache arrow batches.
+ ARROW = 1;
+}
+
+// The UDF execution type/shape.
+message UDFShape {
+ oneof shape {
+ SparkUDFShape spark = 1;
+ }
+}
+
+enum SparkUDFShape {
+ SPARK_UDF_SHAPE_UNSPECIFIED = 0;
+
+ // UDF receives a row with 0+ columns as input
+ // and produces a single, scalar value as output
+ EXPRESSION = 1;
+
+ // UDF receives an iterator to a batch of rows as input and
+ // produces an iterator to a batch of rows as output.
+ MAP_PARTITIONS = 2;
Review Comment:
Deprecated according to the above comment
--
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
Re: [PR] [SPARK-56284] Adding UDF worker specification protobuf definition [spark]
Posted by "hvanhovell (via GitHub)" <gi...@apache.org>.
hvanhovell commented on code in PR #55165:
URL: https://github.com/apache/spark/pull/55165#discussion_r3058790546
##########
udf/worker/proto/src/main/protobuf/worker_spec.proto:
##########
@@ -17,12 +17,194 @@
syntax = "proto3";
+import "common.proto";
+
package org.apache.spark.udf.worker;
option java_package = "org.apache.spark.udf.worker";
option java_multiple_files = true;
-// Placeholder -- full worker specification schema to be added.
-// See design doc: SPIP SPARK-55278.
+///
+/// Worker specification
+///
message UDFWorkerSpecification {
+ WorkerEnvironment environment = 1;
+ WorkerCapabilities capabilities = 2;
+
+ // How to create new workers.
+ // At the moment, only direct creation is supported.
+ // This can be extended with indirect/provisioned creation in the future.
+ oneof worker {
+ DirectWorker direct = 3;
+ }
+}
+
+message WorkerEnvironment {
+ // Callable that is responsible for environment setup.
+ optional ProcessCallable installation = 1;
+ // Callable, which is called to verify that an environment
+ // is suitable to start the UDF worker. This callable
+ // needs to verify that
+ // - The worker code itself is present
+ // - Any needed dependencies are present
+ optional ProcessCallable environment_verification = 2;
+ // Callable, which is invoked after the worker has been terminated.
+ // This can be used to cleanup dependencies or temporary resources.
+ // Be careful not to cleanup resources that could be used by
+ // other workers running in parallel.
+ optional ProcessCallable environment_cleanup = 3;
+}
+
+// Capabilities used for query planning
+message WorkerCapabilities {
+ // The data formats that the worker supports for UDF data in- & output.
+ // Every worker MUST at least support ARROW.
+ //
+ // It is expected that for each UDF execution, the input format
+ // always matches the output format.
+ //
+ // If a worker supports multiple data formats, the engine will select
+ // the most suitable one for each UDF invocation. Which format was chosen
+ // is reported by the engine as part of the UDF protocol's init message.
+ repeated UDFWorkerDataFormat supported_data_formats = 1;
+
+ // Which types of UDFs this worker supports.
+ // This should list all supported Shapes.
+ // The shape of a specific UDF will be communicated
+ // in the initial message of the UDF protocol.
+ //
+ // If an execution for an unsupported UDF type is requested
+ // the query will fail during query planning.
+ repeated UDFShape supported_udf_shapes = 2;
+
+ // Whether multiple, concurrent UDF
+ // connections are supported by this worker
+ // (for example via multi-threading).
+ //
+ // In the first implementation of the engine-side
+ // worker specification, this property will not be used.
+ //
+ // Usage of this property can be enabled in the future if the
+ // engine implements more advanced resource management (TBD).
+ //
+ bool supports_concurrent_udfs = 3;
+
+ // Whether compatible workers may be reused.
+ // If this is not supported, the worker is
+ // terminated after every single UDF invocation.
+ bool supports_reuse = 4;
+
+ // To be extended with UDF chaining, ...
+}
+
+// The worker that will be created to process UDFs
+message DirectWorker {
+ // Blocking callable that is terminated by SIGTERM/SIGKILL
+ ProcessCallable runner = 1;
+
+ UDFWorkerProperties properties = 2;
+}
+
+message UDFWorkerProperties {
+ // Maximum amount of time to wait until the worker can accept connections.
+ //
+ // The engine will use this timeout, if it does not exceed an
+ // engine-configurable maximum time (e.g. 30 seconds).
+ optional int32 initialization_timeout_ms = 1;
+
+ // Used for graceful engine-initiated termination signaled via SIGTERM.
+ // After this time, the worker process should have terminated itself.
+ // Otherwise, the process will be forcefully killed using SIGKILL.
+ //
+ // The engine will use this timeout, if it does not exceed an
+ // engine-configurable maximum time (e.g. 30 seconds).
+ optional int32 graceful_termination_timeout_ms = 2;
+
+ // The connection this [[DirectWorker]] supports. Note that a single
+ // connection is sufficient to run multiple UDFs and (gRPC) services.
+ //
+ // On [[DirectWorker]] creation, connection information
+ // is passed to the callable as a string parameter.
+ // The string format depends on the [[WorkerConnection]]:
+ //
+ // For example, when using TCP, the callable argument will be:
+ // --connection PORT
+ // Here is a concrete example
+ // --connection 8080
+ //
+ // For the format of each specific transport type, see the comments below.
+ WorkerConnection connection = 3;
+}
+
+message WorkerConnection {
+ oneof transport {
+ UnixDomainSocket unix_domain_socket = 1;
+ TcpConnection tcp = 2;
+ }
}
+
+// Communication between the engine and worker
+// is done using a UNIX domain socket.
+//
+// On [[DirectWorker]] creation, a path to a socket
+// to listen on is passed as an argument.
+// Examples:
+// /tmp/channel-uuid.sock
+// /some/system/path/channel-1234.sock
+message UnixDomainSocket {}
+
+// Communication between the engine and worker
+// is done using a localhost TCP connection.
+//
+// On [[DirectWorker]] creation, a PORT
+// is passed as a connection parameter.
+//
+// It is expected that the worker binds to this
+// port on both IPv4 and IPv6 localhost interfaces.
+// E.g. the worker server should be reachable via
+// 127.0.0.1:PORT and [::1]:PORT.
+//
+// Examples:
+// 8080
+// 1234
+message TcpConnection {}
Review Comment:
If this is local only, then name it as such?
Same question as before. I assume you are going to add a port 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
Re: [PR] [SPARK-56284] Adding UDF worker specification protobuf definition [spark]
Posted by "sven-weber-db (via GitHub)" <gi...@apache.org>.
sven-weber-db commented on code in PR #55165:
URL: https://github.com/apache/spark/pull/55165#discussion_r3072628114
##########
udf/worker/proto/src/main/protobuf/worker_spec.proto:
##########
@@ -17,12 +17,194 @@
syntax = "proto3";
+import "common.proto";
+
package org.apache.spark.udf.worker;
option java_package = "org.apache.spark.udf.worker";
option java_multiple_files = true;
-// Placeholder -- full worker specification schema to be added.
-// See design doc: SPIP SPARK-55278.
+///
+/// Worker specification
+///
message UDFWorkerSpecification {
+ WorkerEnvironment environment = 1;
+ WorkerCapabilities capabilities = 2;
+
+ // How to create new workers.
+ // At the moment, only direct creation is supported.
+ // This can be extended with indirect/provisioned creation in the future.
+ oneof worker {
+ DirectWorker direct = 3;
+ }
+}
+
+message WorkerEnvironment {
Review Comment:
Good point. I added additional documentation and a small flowchart showing which callables will be invoked when and in what order.
--
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
Re: [PR] [SPARK-56284] Adding UDF worker specification protobuf definition [spark]
Posted by "hvanhovell (via GitHub)" <gi...@apache.org>.
hvanhovell commented on code in PR #55165:
URL: https://github.com/apache/spark/pull/55165#discussion_r3075021135
##########
udf/worker/proto/src/main/protobuf/worker_spec.proto:
##########
@@ -17,12 +17,194 @@
syntax = "proto3";
+import "common.proto";
+
package org.apache.spark.udf.worker;
option java_package = "org.apache.spark.udf.worker";
option java_multiple_files = true;
-// Placeholder -- full worker specification schema to be added.
-// See design doc: SPIP SPARK-55278.
+///
+/// Worker specification
+///
message UDFWorkerSpecification {
+ WorkerEnvironment environment = 1;
Review Comment:
Ok, let me ask a different different question then: Are there going to be different ways in which a customer can setup a worker environment? If there are then we should at least make this a oneOf... (which we can do later on as well).
--
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
Re: [PR] [SPARK-56284] Adding UDF worker specification protobuf definition [spark]
Posted by "hvanhovell (via GitHub)" <gi...@apache.org>.
hvanhovell commented on PR #55165:
URL: https://github.com/apache/spark/pull/55165#issuecomment-4238741153
Merging this to master. We can address further concerns in follow-ups.
--
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
Re: [PR] [SPARK-56284] Adding UDF worker specification protobuf definition [spark]
Posted by "hvanhovell (via GitHub)" <gi...@apache.org>.
hvanhovell commented on code in PR #55165:
URL: https://github.com/apache/spark/pull/55165#discussion_r3058771445
##########
udf/worker/proto/src/main/protobuf/common.proto:
##########
@@ -0,0 +1,51 @@
+/*
+ * 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 org.apache.spark.udf.worker;
+
+option java_package = "org.apache.spark.udf.worker";
+
+option java_multiple_files = true;
+
+// The UDF in & output data format.
+enum UDFWorkerDataFormat {
+ UDF_WORKER_DATA_FORMAT_UNSPECIFIED = 0;
+
+ // The worker accepts and produces Apache arrow batches.
+ ARROW = 1;
+}
+
+// The UDF execution type/shape.
+message UDFShape {
+ oneof shape {
+ SparkUDFShape spark = 1;
+ }
+}
+
+enum SparkUDFShape {
+ SPARK_UDF_SHAPE_UNSPECIFIED = 0;
+
+ // UDF receives a row with 0+ columns as input
+ // and produces a single, scalar value as output
+ EXPRESSION = 1;
+
+ // UDF receives an iterator to a batch of rows as input and
+ // produces an iterator to a batch of rows as output.
+ MAP_PARTITIONS = 2;
Review Comment:
`MANY_TO_MANY`?
--
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
Re: [PR] [SPARK-56284] Adding UDF worker specification protobuf definition [spark]
Posted by "hvanhovell (via GitHub)" <gi...@apache.org>.
hvanhovell commented on code in PR #55165:
URL: https://github.com/apache/spark/pull/55165#discussion_r3073906277
##########
udf/worker/proto/src/main/protobuf/worker_spec.proto:
##########
@@ -17,12 +17,272 @@
syntax = "proto3";
+import "common.proto";
+
package org.apache.spark.udf.worker;
option java_package = "org.apache.spark.udf.worker";
option java_multiple_files = true;
-// Placeholder -- full worker specification schema to be added.
-// See design doc: SPIP SPARK-55278.
+///
+/// Worker specification
+///
message UDFWorkerSpecification {
+ // (Required)
+ WorkerEnvironment environment = 1;
+ // (Required)
+ WorkerCapabilities capabilities = 2;
+
+ // How to create new workers.
+ // At the moment, only direct creation is supported.
+ // This can be extended with indirect/provisioned creation in the future.
+ //
+ // (Required)
+ oneof worker {
+ DirectWorker direct = 3;
+ }
+}
+
+// Set of callables that can be used to setup, verify,
+// and cleanup the worker environment before the worker
+// callable is invoked.
+//
+// These callables should be specified if the worker
+// setup cannot be assumed to be pre-installed on the
+// target cluster.
+//
+// If defined, the below callables will be invoked once
+// per Spark executor in the following order (top -> bottom):
+//
+// ┌──────────────┐
+// │ │
+// │ Verification │
+// │ │
+// └──┬────────┬──┘
+// │ │
+// Failed │ │ Succeeded
+// │ │
+// ┌──────▼───────┐│
+// │ ││
+// │ Installation ││
+// │ ││
+// └──────┬───────┘│
+// │ │
+// ▼ ▼
+// ...
+// UDF worker creation
+// ...
+// │
+// │
+// ┌────▼────┐
+// │ │
+// │ Cleanup │
+// │ │
+// └─────────┘
+//
+// All scripts are optional.
+// However, if a verification script is supplied, an installation
+// script needs to be supplied as well.
+//
+message WorkerEnvironment {
+ // A callable that prepares cluster nodes by installing
+ // the required runtime, dependencies, and UDF worker binaries
+ // at the expected locations. This step ensures that
+ // the worker starting callable
+ // can subsequently locate and launch the worker process.
+ //
+ // (Required if environment_verification is given)
+ optional ProcessCallable installation = 1;
Review Comment:
Document that installation failure is fatal?
--
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
Re: [PR] [SPARK-56284] Adding UDF worker specification protobuf definition [spark]
Posted by "hvanhovell (via GitHub)" <gi...@apache.org>.
hvanhovell commented on code in PR #55165:
URL: https://github.com/apache/spark/pull/55165#discussion_r3058883838
##########
udf/worker/proto/src/main/protobuf/worker_spec.proto:
##########
@@ -17,12 +17,194 @@
syntax = "proto3";
+import "common.proto";
+
package org.apache.spark.udf.worker;
option java_package = "org.apache.spark.udf.worker";
option java_multiple_files = true;
-// Placeholder -- full worker specification schema to be added.
-// See design doc: SPIP SPARK-55278.
+///
+/// Worker specification
+///
message UDFWorkerSpecification {
+ WorkerEnvironment environment = 1;
+ WorkerCapabilities capabilities = 2;
+
+ // How to create new workers.
+ // At the moment, only direct creation is supported.
+ // This can be extended with indirect/provisioned creation in the future.
+ oneof worker {
+ DirectWorker direct = 3;
+ }
+}
+
+message WorkerEnvironment {
+ // Callable that is responsible for environment setup.
+ optional ProcessCallable installation = 1;
+ // Callable, which is called to verify that an environment
+ // is suitable to start the UDF worker. This callable
+ // needs to verify that
+ // - The worker code itself is present
+ // - Any needed dependencies are present
+ optional ProcessCallable environment_verification = 2;
+ // Callable, which is invoked after the worker has been terminated.
+ // This can be used to cleanup dependencies or temporary resources.
+ // Be careful not to cleanup resources that could be used by
+ // other workers running in parallel.
+ optional ProcessCallable environment_cleanup = 3;
+}
+
+// Capabilities used for query planning
+message WorkerCapabilities {
Review Comment:
IMO it would be a good idea to make this have a somewhat self describing name.
--
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
Re: [PR] [SPARK-56284] Adding UDF worker specification protobuf definition [spark]
Posted by "hvanhovell (via GitHub)" <gi...@apache.org>.
hvanhovell commented on code in PR #55165:
URL: https://github.com/apache/spark/pull/55165#discussion_r3058862779
##########
udf/worker/proto/src/main/protobuf/worker_spec.proto:
##########
@@ -17,12 +17,194 @@
syntax = "proto3";
+import "common.proto";
+
package org.apache.spark.udf.worker;
option java_package = "org.apache.spark.udf.worker";
option java_multiple_files = true;
-// Placeholder -- full worker specification schema to be added.
-// See design doc: SPIP SPARK-55278.
+///
+/// Worker specification
+///
message UDFWorkerSpecification {
+ WorkerEnvironment environment = 1;
Review Comment:
WorkerEnvironment looks like it is in the wrong place. It uses a lot of ProcessCallables which only seem to make sense for DirectWorker. Why not make it part of the direct worker to begin with?
--
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
Re: [PR] [SPARK-56284] Adding UDF worker specification protobuf definition [spark]
Posted by "cloud-fan (via GitHub)" <gi...@apache.org>.
cloud-fan commented on code in PR #55165:
URL: https://github.com/apache/spark/pull/55165#discussion_r3045642972
##########
udf/worker/proto/src/main/protobuf/common.proto:
##########
@@ -0,0 +1,51 @@
+/*
+ * 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 org.apache.spark.udf.worker;
+
+option java_package = "org.apache.spark.udf.worker";
+
+option java_multiple_files = true;
+
+// The UDF in & output data format.
+enum UDFWorkerDataFormat {
+ UDF_WORKER_DATA_FORMAT_UNSPECIFIED = 0;
+
+ // The worker accepts and produces Apache arrow batches.
+ ARROW = 1;
+}
+
+// The UDF execution type/shape.
+message UDFShape {
+ oneof shape {
+ SparkUDFShapes spark = 1;
+ }
+}
+
+enum SparkUDFShapes {
+ SPARK_UDF_SHAPE_UNSPECIFIED = 0;
+
+ // UDF receives a row with 0+ columns as input
+ // and produces a single, scalar value as output
+ EXPRESSION = 1;
+
+ // UDF receives a iterator to batch of rows as input and
+ // produces iterator to a batch of rows as output.
+ MAP_PARTITIONS = 2;
Review Comment:
Grammar — "a iterator" and missing article:
```suggestion
// UDF receives an iterator to a batch of rows as input and
// produces an iterator to a batch of rows as output.
```
##########
udf/worker/proto/src/main/protobuf/worker_spec.proto:
##########
@@ -17,12 +17,186 @@
syntax = "proto3";
+import "common.proto";
+
package org.apache.spark.udf.worker;
option java_package = "org.apache.spark.udf.worker";
option java_multiple_files = true;
-// Placeholder -- full worker specification schema to be added.
-// See design doc: SPIP SPARK-55278.
+///
+/// Worker specification
+///
message UDFWorkerSpecification {
+ WorkerEnvironment environment = 1;
+ WorkerCapabilities capabilities = 2;
+
+ // How to create new workers.
+ // At the moment, only direct creation is supported.
+ // This can be extended with indirect/provisioned creation in the future.
+ oneof worker {
+ DirectWorker direct = 3;
+ }
+}
+
+message WorkerEnvironment {
+ // Callable that is responsible for environment setup.
+ optional ProcessCallable installation = 1;
+ // Callable, which is called to verify that an environment
+ // is suitable to start the UDF worker. This callable
+ // needs to verify that
+ // - The worker code itself is present
+ // - Any needed dependencies are present
+ optional ProcessCallable environment_verification = 2;
+ // Callable, which is invoked after the worker has been terminated.
+ // This can be used to cleanup dependencies or temporary resources.
+ // Be careful not to cleanup resources that could be used by
+ // other workers running in parallel.
+ optional ProcessCallable environment_cleanup = 3;
+}
+
+// Capabilities used for query planning
+message WorkerCapabilities {
+ // The data formats that the worker supports for UDF data in- & output.
+ // Every worker MUST at least support ARROW.
+ //
+ // It is expected that for each UDF execution, the input format
+ // always matches the output format.
+ //
+ // If a worker supports multiple data formats, the engine will select
+ // the most suitable one for each UDF invocation. Which format was chosen
+ // is reported by the engine as part of the UDF protocol's init message.
+ repeated UDFWorkerDataFormat supported_data_formats = 1;
+
+ // Which types of UDFs this worker supports.
+ // This should list all supported Shapes.
+ // Of which shape a specific UDF is will be communicated
Review Comment:
Awkward phrasing:
```suggestion
// The shape of a specific UDF will be communicated
```
##########
udf/worker/proto/src/main/protobuf/worker_spec.proto:
##########
@@ -17,12 +17,186 @@
syntax = "proto3";
+import "common.proto";
+
package org.apache.spark.udf.worker;
option java_package = "org.apache.spark.udf.worker";
option java_multiple_files = true;
-// Placeholder -- full worker specification schema to be added.
-// See design doc: SPIP SPARK-55278.
+///
+/// Worker specification
+///
message UDFWorkerSpecification {
+ WorkerEnvironment environment = 1;
+ WorkerCapabilities capabilities = 2;
+
+ // How to create new workers.
+ // At the moment, only direct creation is supported.
+ // This can be extended with indirect/provisioned creation in the future.
+ oneof worker {
+ DirectWorker direct = 3;
+ }
+}
+
+message WorkerEnvironment {
+ // Callable that is responsible for environment setup.
+ optional ProcessCallable installation = 1;
+ // Callable, which is called to verify that an environment
+ // is suitable to start the UDF worker. This callable
+ // needs to verify that
+ // - The worker code itself is present
+ // - Any needed dependencies are present
+ optional ProcessCallable environment_verification = 2;
+ // Callable, which is invoked after the worker has been terminated.
+ // This can be used to cleanup dependencies or temporary resources.
+ // Be careful not to cleanup resources that could be used by
+ // other workers running in parallel.
+ optional ProcessCallable environment_cleanup = 3;
+}
+
+// Capabilities used for query planning
+message WorkerCapabilities {
+ // The data formats that the worker supports for UDF data in- & output.
+ // Every worker MUST at least support ARROW.
+ //
+ // It is expected that for each UDF execution, the input format
+ // always matches the output format.
+ //
+ // If a worker supports multiple data formats, the engine will select
+ // the most suitable one for each UDF invocation. Which format was chosen
+ // is reported by the engine as part of the UDF protocol's init message.
+ repeated UDFWorkerDataFormat supported_data_formats = 1;
+
+ // Which types of UDFs this worker supports.
+ // This should list all supported Shapes.
+ // Of which shape a specific UDF is will be communicated
+ // in the initial message of the UDF protocol.
+ //
+ // If a execution for an unsupported UDF type is requested
+ // the query will fail during query planning.
+ repeated UDFShape supported_udf_shapes = 2;
+
+ // Whether multiple, concurrent UDF
+ // connections are supported by this worker
+ // (for example via multi-threading).
+ //
+ // In the first implementation of the engine-side
+ // worker specification, this property will not be used.
+ //
+ // Usage of this property, can be enabled in the future if the
+ // engine implements more advanced resource management (Tbd).
+ //
+ bool supports_concurrent_udfs = 3;
+
+ // Whether compatible workers may be reused.
+ // If this is not supported, the worker is
+ // terminated after every single UDF invocation.
+ bool supports_reuse = 4;
+
+ // To be extended with UDF chaining, ...
+}
+
+// The worker that will be created to process UDFs
+message DirectWorker {
+ // Blocking callable that is terminated by SIGTERM/SIGKILL
+ ProcessCallable runner = 1;
+
+ UDFWorkerProperties properties = 2;
+}
+
+message UDFWorkerProperties {
+ // Maximum amount of time to wait until the worker can accept connections.
+ //
+ // The engine will use this timeout, if it does not exceed a
+ // engine-configurable maximum time (e.g. 30 seconds).
+ optional int32 initialization_timeout_ms = 1;
+
+ // Used for graceful engine-initiated termination signaled via SIGTERM.
+ // After this time, the worker process should have terminated itself.
+ // Otherwise, the process will be forcefully killed using SIGKILL.
+ //
+ // The engine will use this timeout, if it does not exceed a
+ // engine-configurable maximum time (e.g. 30 seconds).
+ optional int32 graceful_termination_timeout_ms = 2;
+
+ // The connection this [[UDFWorker]] supports. Note that a single
Review Comment:
`[[UDFWorker]]` is not defined anywhere — no proto message, no Scala/Java class. The same dangling reference appears at lines 149 and 159. The closest entity is `DirectWorker` (line 101). Should these reference `DirectWorker`, or is `UDFWorker` a planned type not yet introduced?
##########
udf/worker/proto/src/main/protobuf/worker_spec.proto:
##########
@@ -17,12 +17,186 @@
syntax = "proto3";
+import "common.proto";
+
package org.apache.spark.udf.worker;
option java_package = "org.apache.spark.udf.worker";
option java_multiple_files = true;
-// Placeholder -- full worker specification schema to be added.
-// See design doc: SPIP SPARK-55278.
+///
+/// Worker specification
+///
message UDFWorkerSpecification {
+ WorkerEnvironment environment = 1;
+ WorkerCapabilities capabilities = 2;
+
+ // How to create new workers.
+ // At the moment, only direct creation is supported.
+ // This can be extended with indirect/provisioned creation in the future.
+ oneof worker {
+ DirectWorker direct = 3;
+ }
+}
+
+message WorkerEnvironment {
+ // Callable that is responsible for environment setup.
+ optional ProcessCallable installation = 1;
+ // Callable, which is called to verify that an environment
+ // is suitable to start the UDF worker. This callable
+ // needs to verify that
+ // - The worker code itself is present
+ // - Any needed dependencies are present
+ optional ProcessCallable environment_verification = 2;
+ // Callable, which is invoked after the worker has been terminated.
+ // This can be used to cleanup dependencies or temporary resources.
+ // Be careful not to cleanup resources that could be used by
+ // other workers running in parallel.
+ optional ProcessCallable environment_cleanup = 3;
+}
+
+// Capabilities used for query planning
+message WorkerCapabilities {
+ // The data formats that the worker supports for UDF data in- & output.
+ // Every worker MUST at least support ARROW.
+ //
+ // It is expected that for each UDF execution, the input format
+ // always matches the output format.
+ //
+ // If a worker supports multiple data formats, the engine will select
+ // the most suitable one for each UDF invocation. Which format was chosen
+ // is reported by the engine as part of the UDF protocol's init message.
+ repeated UDFWorkerDataFormat supported_data_formats = 1;
+
+ // Which types of UDFs this worker supports.
+ // This should list all supported Shapes.
+ // Of which shape a specific UDF is will be communicated
+ // in the initial message of the UDF protocol.
+ //
+ // If a execution for an unsupported UDF type is requested
+ // the query will fail during query planning.
+ repeated UDFShape supported_udf_shapes = 2;
+
+ // Whether multiple, concurrent UDF
+ // connections are supported by this worker
+ // (for example via multi-threading).
+ //
+ // In the first implementation of the engine-side
+ // worker specification, this property will not be used.
+ //
+ // Usage of this property, can be enabled in the future if the
+ // engine implements more advanced resource management (Tbd).
+ //
+ bool supports_concurrent_udfs = 3;
+
+ // Whether compatible workers may be reused.
+ // If this is not supported, the worker is
+ // terminated after every single UDF invocation.
+ bool supports_reuse = 4;
+
+ // To be extended with UDF chaining, ...
+}
+
+// The worker that will be created to process UDFs
+message DirectWorker {
+ // Blocking callable that is terminated by SIGTERM/SIGKILL
+ ProcessCallable runner = 1;
+
+ UDFWorkerProperties properties = 2;
+}
+
+message UDFWorkerProperties {
+ // Maximum amount of time to wait until the worker can accept connections.
+ //
+ // The engine will use this timeout, if it does not exceed a
+ // engine-configurable maximum time (e.g. 30 seconds).
+ optional int32 initialization_timeout_ms = 1;
+
+ // Used for graceful engine-initiated termination signaled via SIGTERM.
+ // After this time, the worker process should have terminated itself.
+ // Otherwise, the process will be forcefully killed using SIGKILL.
+ //
+ // The engine will use this timeout, if it does not exceed a
+ // engine-configurable maximum time (e.g. 30 seconds).
+ optional int32 graceful_termination_timeout_ms = 2;
+
+ // The connection this [[UDFWorker]] supports. Note that a single
+ // connection is sufficient to run multiple UDFs and (gRPC) services.
+ //
+ // On [[UDFWorker]] creation, connection information
+ // is passed to the callable as a string parameter.
+ // The string format depends on the [[WorkerConnection]]:
+ //
+ // For example, when using TCP, the callable argument will be:
+ // --connection PORT
+ // Here is a concrete example
+ // --connection 8080
+ //
+ // For the format of each specific transport type, see the comments below.
+ WorkerConnection connection = 3;
+}
+
+message WorkerConnection {
+ oneof transport {
+ UnixDomainSocket unix_domain_socket = 1;
+ TcpConnection tcp = 2;
+ }
}
+
+ // Communication between the engine and worker
+// is done using a UNIX domain socket.
+//
+// On [[UDFWorker]] creation, a path to a socket
+// to listen on is passed as a argument.
+// Examples:
+// /tmp/channel-uuid.sock
+// /some/system/path/channel-1234.sock
+message UnixDomainSocket {}
+
+// Communication between the engine and worker
+// is done using a localhost TCP connection.
+//
+// On [[UDFWorker]] creation, a PORT
+// is passed as a connection parameter.
+//
+// It is expected that the worker binds to this
+// port on both IPv4 and IPv6 localhost interfaces.
+// E.g. the worker server should be reachable via
+// 127.0.0.1:PORT and [::1]:PORT.
+//
+// Examples:
+// 8080
+// 1234
+message TcpConnection {}
+
+message ProcessCallable {
+ // Executable to invoke.
+ // Examples:
+ // ["python3", "-m"]
+ // ["worker.bin"]
+ // ["java", "worker.java"]
+ // ["bin/bash", "-c"]
+ // This executable should be blocking, until the task is finished.
+ // Termination is requested via a SIGTERM signal.
+ //
+ // Success/Failure can be indicated via exit codes:
+ // Exit code 0 -> Success
+ // Exit code != 0 -> Failure
+ repeated string command = 1;
+
+ // Arguments passed directly to the executable.
+ // Examples:
+ // ["udf_worker.py"]
+ // [""]
+ // ["--max_concurrency", "5"]
+ // ["\"echo 'Test'\""]
+ //
+ // Every executable will ALWAYS receive a
+ // --id argument. This argument CANNOT be part of the below list of arguments.
Review Comment:
The `--id` argument is explicitly reserved here ("CANNOT be part of the below list of arguments"), but `--connection` (injected by the engine per lines 130–134) has no such restriction documented. A user including `--connection` in their `arguments` would conflict with the engine-injected value. Consider adding the same reservation for `--connection`.
##########
udf/worker/proto/src/main/protobuf/worker_spec.proto:
##########
@@ -17,12 +17,186 @@
syntax = "proto3";
+import "common.proto";
+
package org.apache.spark.udf.worker;
option java_package = "org.apache.spark.udf.worker";
option java_multiple_files = true;
-// Placeholder -- full worker specification schema to be added.
-// See design doc: SPIP SPARK-55278.
+///
+/// Worker specification
+///
message UDFWorkerSpecification {
+ WorkerEnvironment environment = 1;
+ WorkerCapabilities capabilities = 2;
+
+ // How to create new workers.
+ // At the moment, only direct creation is supported.
+ // This can be extended with indirect/provisioned creation in the future.
+ oneof worker {
+ DirectWorker direct = 3;
+ }
+}
+
+message WorkerEnvironment {
+ // Callable that is responsible for environment setup.
+ optional ProcessCallable installation = 1;
+ // Callable, which is called to verify that an environment
+ // is suitable to start the UDF worker. This callable
+ // needs to verify that
+ // - The worker code itself is present
+ // - Any needed dependencies are present
+ optional ProcessCallable environment_verification = 2;
+ // Callable, which is invoked after the worker has been terminated.
+ // This can be used to cleanup dependencies or temporary resources.
+ // Be careful not to cleanup resources that could be used by
+ // other workers running in parallel.
+ optional ProcessCallable environment_cleanup = 3;
+}
+
+// Capabilities used for query planning
+message WorkerCapabilities {
+ // The data formats that the worker supports for UDF data in- & output.
+ // Every worker MUST at least support ARROW.
+ //
+ // It is expected that for each UDF execution, the input format
+ // always matches the output format.
+ //
+ // If a worker supports multiple data formats, the engine will select
+ // the most suitable one for each UDF invocation. Which format was chosen
+ // is reported by the engine as part of the UDF protocol's init message.
+ repeated UDFWorkerDataFormat supported_data_formats = 1;
+
+ // Which types of UDFs this worker supports.
+ // This should list all supported Shapes.
+ // Of which shape a specific UDF is will be communicated
+ // in the initial message of the UDF protocol.
+ //
+ // If a execution for an unsupported UDF type is requested
+ // the query will fail during query planning.
+ repeated UDFShape supported_udf_shapes = 2;
+
+ // Whether multiple, concurrent UDF
+ // connections are supported by this worker
+ // (for example via multi-threading).
+ //
+ // In the first implementation of the engine-side
+ // worker specification, this property will not be used.
+ //
+ // Usage of this property, can be enabled in the future if the
+ // engine implements more advanced resource management (Tbd).
+ //
+ bool supports_concurrent_udfs = 3;
+
+ // Whether compatible workers may be reused.
+ // If this is not supported, the worker is
+ // terminated after every single UDF invocation.
+ bool supports_reuse = 4;
+
+ // To be extended with UDF chaining, ...
+}
+
+// The worker that will be created to process UDFs
+message DirectWorker {
+ // Blocking callable that is terminated by SIGTERM/SIGKILL
+ ProcessCallable runner = 1;
+
+ UDFWorkerProperties properties = 2;
+}
+
+message UDFWorkerProperties {
+ // Maximum amount of time to wait until the worker can accept connections.
+ //
+ // The engine will use this timeout, if it does not exceed a
+ // engine-configurable maximum time (e.g. 30 seconds).
+ optional int32 initialization_timeout_ms = 1;
+
+ // Used for graceful engine-initiated termination signaled via SIGTERM.
+ // After this time, the worker process should have terminated itself.
+ // Otherwise, the process will be forcefully killed using SIGKILL.
+ //
+ // The engine will use this timeout, if it does not exceed a
+ // engine-configurable maximum time (e.g. 30 seconds).
+ optional int32 graceful_termination_timeout_ms = 2;
+
+ // The connection this [[UDFWorker]] supports. Note that a single
+ // connection is sufficient to run multiple UDFs and (gRPC) services.
+ //
+ // On [[UDFWorker]] creation, connection information
+ // is passed to the callable as a string parameter.
+ // The string format depends on the [[WorkerConnection]]:
+ //
+ // For example, when using TCP, the callable argument will be:
+ // --connection PORT
+ // Here is a concrete example
+ // --connection 8080
+ //
+ // For the format of each specific transport type, see the comments below.
+ WorkerConnection connection = 3;
+}
+
+message WorkerConnection {
+ oneof transport {
+ UnixDomainSocket unix_domain_socket = 1;
+ TcpConnection tcp = 2;
+ }
}
+
+ // Communication between the engine and worker
Review Comment:
Leading space before `//` — inconsistent with all other message-level comments:
```suggestion
// Communication between the engine and worker
```
##########
udf/worker/proto/src/main/protobuf/worker_spec.proto:
##########
@@ -17,12 +17,186 @@
syntax = "proto3";
+import "common.proto";
+
package org.apache.spark.udf.worker;
option java_package = "org.apache.spark.udf.worker";
option java_multiple_files = true;
-// Placeholder -- full worker specification schema to be added.
-// See design doc: SPIP SPARK-55278.
+///
+/// Worker specification
+///
message UDFWorkerSpecification {
+ WorkerEnvironment environment = 1;
+ WorkerCapabilities capabilities = 2;
+
+ // How to create new workers.
+ // At the moment, only direct creation is supported.
+ // This can be extended with indirect/provisioned creation in the future.
+ oneof worker {
+ DirectWorker direct = 3;
+ }
+}
+
+message WorkerEnvironment {
+ // Callable that is responsible for environment setup.
+ optional ProcessCallable installation = 1;
+ // Callable, which is called to verify that an environment
+ // is suitable to start the UDF worker. This callable
+ // needs to verify that
+ // - The worker code itself is present
+ // - Any needed dependencies are present
+ optional ProcessCallable environment_verification = 2;
+ // Callable, which is invoked after the worker has been terminated.
+ // This can be used to cleanup dependencies or temporary resources.
+ // Be careful not to cleanup resources that could be used by
+ // other workers running in parallel.
+ optional ProcessCallable environment_cleanup = 3;
+}
+
+// Capabilities used for query planning
+message WorkerCapabilities {
+ // The data formats that the worker supports for UDF data in- & output.
+ // Every worker MUST at least support ARROW.
+ //
+ // It is expected that for each UDF execution, the input format
+ // always matches the output format.
+ //
+ // If a worker supports multiple data formats, the engine will select
+ // the most suitable one for each UDF invocation. Which format was chosen
+ // is reported by the engine as part of the UDF protocol's init message.
+ repeated UDFWorkerDataFormat supported_data_formats = 1;
+
+ // Which types of UDFs this worker supports.
+ // This should list all supported Shapes.
+ // Of which shape a specific UDF is will be communicated
+ // in the initial message of the UDF protocol.
+ //
+ // If a execution for an unsupported UDF type is requested
+ // the query will fail during query planning.
+ repeated UDFShape supported_udf_shapes = 2;
+
+ // Whether multiple, concurrent UDF
+ // connections are supported by this worker
+ // (for example via multi-threading).
+ //
+ // In the first implementation of the engine-side
+ // worker specification, this property will not be used.
+ //
+ // Usage of this property, can be enabled in the future if the
+ // engine implements more advanced resource management (Tbd).
+ //
+ bool supports_concurrent_udfs = 3;
+
+ // Whether compatible workers may be reused.
+ // If this is not supported, the worker is
+ // terminated after every single UDF invocation.
+ bool supports_reuse = 4;
+
+ // To be extended with UDF chaining, ...
+}
+
+// The worker that will be created to process UDFs
+message DirectWorker {
+ // Blocking callable that is terminated by SIGTERM/SIGKILL
+ ProcessCallable runner = 1;
+
+ UDFWorkerProperties properties = 2;
+}
+
+message UDFWorkerProperties {
+ // Maximum amount of time to wait until the worker can accept connections.
+ //
+ // The engine will use this timeout, if it does not exceed a
+ // engine-configurable maximum time (e.g. 30 seconds).
Review Comment:
Same issue at line 119.
```suggestion
// The engine will use this timeout, if it does not exceed an
```
##########
udf/worker/proto/src/main/protobuf/worker_spec.proto:
##########
@@ -17,12 +17,186 @@
syntax = "proto3";
+import "common.proto";
+
package org.apache.spark.udf.worker;
option java_package = "org.apache.spark.udf.worker";
option java_multiple_files = true;
-// Placeholder -- full worker specification schema to be added.
-// See design doc: SPIP SPARK-55278.
+///
+/// Worker specification
+///
message UDFWorkerSpecification {
+ WorkerEnvironment environment = 1;
+ WorkerCapabilities capabilities = 2;
+
+ // How to create new workers.
+ // At the moment, only direct creation is supported.
+ // This can be extended with indirect/provisioned creation in the future.
+ oneof worker {
+ DirectWorker direct = 3;
+ }
+}
+
+message WorkerEnvironment {
+ // Callable that is responsible for environment setup.
+ optional ProcessCallable installation = 1;
+ // Callable, which is called to verify that an environment
+ // is suitable to start the UDF worker. This callable
+ // needs to verify that
+ // - The worker code itself is present
+ // - Any needed dependencies are present
+ optional ProcessCallable environment_verification = 2;
+ // Callable, which is invoked after the worker has been terminated.
+ // This can be used to cleanup dependencies or temporary resources.
+ // Be careful not to cleanup resources that could be used by
+ // other workers running in parallel.
+ optional ProcessCallable environment_cleanup = 3;
+}
+
+// Capabilities used for query planning
+message WorkerCapabilities {
+ // The data formats that the worker supports for UDF data in- & output.
+ // Every worker MUST at least support ARROW.
+ //
+ // It is expected that for each UDF execution, the input format
+ // always matches the output format.
+ //
+ // If a worker supports multiple data formats, the engine will select
+ // the most suitable one for each UDF invocation. Which format was chosen
+ // is reported by the engine as part of the UDF protocol's init message.
+ repeated UDFWorkerDataFormat supported_data_formats = 1;
+
+ // Which types of UDFs this worker supports.
+ // This should list all supported Shapes.
+ // Of which shape a specific UDF is will be communicated
+ // in the initial message of the UDF protocol.
+ //
+ // If a execution for an unsupported UDF type is requested
Review Comment:
```suggestion
// If an execution for an unsupported UDF type is requested
```
##########
udf/worker/proto/src/main/protobuf/worker_spec.proto:
##########
@@ -17,12 +17,186 @@
syntax = "proto3";
+import "common.proto";
+
package org.apache.spark.udf.worker;
option java_package = "org.apache.spark.udf.worker";
option java_multiple_files = true;
-// Placeholder -- full worker specification schema to be added.
-// See design doc: SPIP SPARK-55278.
+///
+/// Worker specification
+///
message UDFWorkerSpecification {
+ WorkerEnvironment environment = 1;
+ WorkerCapabilities capabilities = 2;
+
+ // How to create new workers.
+ // At the moment, only direct creation is supported.
+ // This can be extended with indirect/provisioned creation in the future.
+ oneof worker {
+ DirectWorker direct = 3;
+ }
+}
+
+message WorkerEnvironment {
+ // Callable that is responsible for environment setup.
+ optional ProcessCallable installation = 1;
+ // Callable, which is called to verify that an environment
+ // is suitable to start the UDF worker. This callable
+ // needs to verify that
+ // - The worker code itself is present
+ // - Any needed dependencies are present
+ optional ProcessCallable environment_verification = 2;
+ // Callable, which is invoked after the worker has been terminated.
+ // This can be used to cleanup dependencies or temporary resources.
+ // Be careful not to cleanup resources that could be used by
+ // other workers running in parallel.
+ optional ProcessCallable environment_cleanup = 3;
+}
+
+// Capabilities used for query planning
+message WorkerCapabilities {
+ // The data formats that the worker supports for UDF data in- & output.
+ // Every worker MUST at least support ARROW.
+ //
+ // It is expected that for each UDF execution, the input format
+ // always matches the output format.
+ //
+ // If a worker supports multiple data formats, the engine will select
+ // the most suitable one for each UDF invocation. Which format was chosen
+ // is reported by the engine as part of the UDF protocol's init message.
+ repeated UDFWorkerDataFormat supported_data_formats = 1;
+
+ // Which types of UDFs this worker supports.
+ // This should list all supported Shapes.
+ // Of which shape a specific UDF is will be communicated
+ // in the initial message of the UDF protocol.
+ //
+ // If a execution for an unsupported UDF type is requested
+ // the query will fail during query planning.
+ repeated UDFShape supported_udf_shapes = 2;
+
+ // Whether multiple, concurrent UDF
+ // connections are supported by this worker
+ // (for example via multi-threading).
+ //
+ // In the first implementation of the engine-side
+ // worker specification, this property will not be used.
+ //
+ // Usage of this property, can be enabled in the future if the
+ // engine implements more advanced resource management (Tbd).
+ //
+ bool supports_concurrent_udfs = 3;
+
+ // Whether compatible workers may be reused.
+ // If this is not supported, the worker is
+ // terminated after every single UDF invocation.
+ bool supports_reuse = 4;
+
+ // To be extended with UDF chaining, ...
+}
+
+// The worker that will be created to process UDFs
+message DirectWorker {
+ // Blocking callable that is terminated by SIGTERM/SIGKILL
+ ProcessCallable runner = 1;
+
+ UDFWorkerProperties properties = 2;
+}
+
+message UDFWorkerProperties {
+ // Maximum amount of time to wait until the worker can accept connections.
+ //
+ // The engine will use this timeout, if it does not exceed a
+ // engine-configurable maximum time (e.g. 30 seconds).
+ optional int32 initialization_timeout_ms = 1;
+
+ // Used for graceful engine-initiated termination signaled via SIGTERM.
+ // After this time, the worker process should have terminated itself.
+ // Otherwise, the process will be forcefully killed using SIGKILL.
+ //
+ // The engine will use this timeout, if it does not exceed a
+ // engine-configurable maximum time (e.g. 30 seconds).
+ optional int32 graceful_termination_timeout_ms = 2;
+
+ // The connection this [[UDFWorker]] supports. Note that a single
+ // connection is sufficient to run multiple UDFs and (gRPC) services.
+ //
+ // On [[UDFWorker]] creation, connection information
+ // is passed to the callable as a string parameter.
+ // The string format depends on the [[WorkerConnection]]:
+ //
+ // For example, when using TCP, the callable argument will be:
+ // --connection PORT
+ // Here is a concrete example
+ // --connection 8080
+ //
+ // For the format of each specific transport type, see the comments below.
+ WorkerConnection connection = 3;
+}
+
+message WorkerConnection {
+ oneof transport {
+ UnixDomainSocket unix_domain_socket = 1;
+ TcpConnection tcp = 2;
+ }
}
+
+ // Communication between the engine and worker
+// is done using a UNIX domain socket.
+//
+// On [[UDFWorker]] creation, a path to a socket
+// to listen on is passed as a argument.
+// Examples:
+// /tmp/channel-uuid.sock
+// /some/system/path/channel-1234.sock
+message UnixDomainSocket {}
+
+// Communication between the engine and worker
+// is done using a localhost TCP connection.
+//
+// On [[UDFWorker]] creation, a PORT
+// is passed as a connection parameter.
+//
+// It is expected that the worker binds to this
+// port on both IPv4 and IPv6 localhost interfaces.
+// E.g. the worker server should be reachable via
+// 127.0.0.1:PORT and [::1]:PORT.
+//
+// Examples:
+// 8080
+// 1234
+message TcpConnection {}
+
+message ProcessCallable {
+ // Executable to invoke.
+ // Examples:
+ // ["python3", "-m"]
+ // ["worker.bin"]
+ // ["java", "worker.java"]
+ // ["bin/bash", "-c"]
Review Comment:
Missing leading `/`:
```suggestion
// ["/bin/bash", "-c"]
```
##########
udf/worker/proto/src/main/protobuf/worker_spec.proto:
##########
@@ -17,12 +17,186 @@
syntax = "proto3";
+import "common.proto";
+
package org.apache.spark.udf.worker;
option java_package = "org.apache.spark.udf.worker";
option java_multiple_files = true;
-// Placeholder -- full worker specification schema to be added.
-// See design doc: SPIP SPARK-55278.
+///
+/// Worker specification
+///
message UDFWorkerSpecification {
+ WorkerEnvironment environment = 1;
+ WorkerCapabilities capabilities = 2;
+
+ // How to create new workers.
+ // At the moment, only direct creation is supported.
+ // This can be extended with indirect/provisioned creation in the future.
+ oneof worker {
+ DirectWorker direct = 3;
+ }
+}
+
+message WorkerEnvironment {
+ // Callable that is responsible for environment setup.
+ optional ProcessCallable installation = 1;
+ // Callable, which is called to verify that an environment
+ // is suitable to start the UDF worker. This callable
+ // needs to verify that
+ // - The worker code itself is present
+ // - Any needed dependencies are present
+ optional ProcessCallable environment_verification = 2;
+ // Callable, which is invoked after the worker has been terminated.
+ // This can be used to cleanup dependencies or temporary resources.
+ // Be careful not to cleanup resources that could be used by
+ // other workers running in parallel.
+ optional ProcessCallable environment_cleanup = 3;
+}
+
+// Capabilities used for query planning
+message WorkerCapabilities {
+ // The data formats that the worker supports for UDF data in- & output.
+ // Every worker MUST at least support ARROW.
+ //
+ // It is expected that for each UDF execution, the input format
+ // always matches the output format.
+ //
+ // If a worker supports multiple data formats, the engine will select
+ // the most suitable one for each UDF invocation. Which format was chosen
+ // is reported by the engine as part of the UDF protocol's init message.
+ repeated UDFWorkerDataFormat supported_data_formats = 1;
+
+ // Which types of UDFs this worker supports.
+ // This should list all supported Shapes.
+ // Of which shape a specific UDF is will be communicated
+ // in the initial message of the UDF protocol.
+ //
+ // If a execution for an unsupported UDF type is requested
+ // the query will fail during query planning.
+ repeated UDFShape supported_udf_shapes = 2;
+
+ // Whether multiple, concurrent UDF
+ // connections are supported by this worker
+ // (for example via multi-threading).
+ //
+ // In the first implementation of the engine-side
+ // worker specification, this property will not be used.
+ //
+ // Usage of this property, can be enabled in the future if the
+ // engine implements more advanced resource management (Tbd).
+ //
+ bool supports_concurrent_udfs = 3;
+
+ // Whether compatible workers may be reused.
+ // If this is not supported, the worker is
+ // terminated after every single UDF invocation.
+ bool supports_reuse = 4;
+
+ // To be extended with UDF chaining, ...
+}
+
+// The worker that will be created to process UDFs
+message DirectWorker {
+ // Blocking callable that is terminated by SIGTERM/SIGKILL
+ ProcessCallable runner = 1;
+
+ UDFWorkerProperties properties = 2;
+}
+
+message UDFWorkerProperties {
+ // Maximum amount of time to wait until the worker can accept connections.
+ //
+ // The engine will use this timeout, if it does not exceed a
+ // engine-configurable maximum time (e.g. 30 seconds).
+ optional int32 initialization_timeout_ms = 1;
+
+ // Used for graceful engine-initiated termination signaled via SIGTERM.
+ // After this time, the worker process should have terminated itself.
+ // Otherwise, the process will be forcefully killed using SIGKILL.
+ //
+ // The engine will use this timeout, if it does not exceed a
+ // engine-configurable maximum time (e.g. 30 seconds).
+ optional int32 graceful_termination_timeout_ms = 2;
+
+ // The connection this [[UDFWorker]] supports. Note that a single
+ // connection is sufficient to run multiple UDFs and (gRPC) services.
+ //
+ // On [[UDFWorker]] creation, connection information
+ // is passed to the callable as a string parameter.
+ // The string format depends on the [[WorkerConnection]]:
+ //
+ // For example, when using TCP, the callable argument will be:
+ // --connection PORT
+ // Here is a concrete example
+ // --connection 8080
+ //
+ // For the format of each specific transport type, see the comments below.
+ WorkerConnection connection = 3;
+}
+
+message WorkerConnection {
+ oneof transport {
+ UnixDomainSocket unix_domain_socket = 1;
+ TcpConnection tcp = 2;
+ }
}
+
+ // Communication between the engine and worker
+// is done using a UNIX domain socket.
+//
+// On [[UDFWorker]] creation, a path to a socket
+// to listen on is passed as a argument.
Review Comment:
```suggestion
// to listen on is passed as an argument.
```
##########
udf/worker/proto/src/main/protobuf/worker_spec.proto:
##########
@@ -17,12 +17,186 @@
syntax = "proto3";
+import "common.proto";
+
package org.apache.spark.udf.worker;
option java_package = "org.apache.spark.udf.worker";
option java_multiple_files = true;
-// Placeholder -- full worker specification schema to be added.
-// See design doc: SPIP SPARK-55278.
+///
+/// Worker specification
+///
message UDFWorkerSpecification {
+ WorkerEnvironment environment = 1;
+ WorkerCapabilities capabilities = 2;
+
+ // How to create new workers.
+ // At the moment, only direct creation is supported.
+ // This can be extended with indirect/provisioned creation in the future.
+ oneof worker {
+ DirectWorker direct = 3;
+ }
+}
+
+message WorkerEnvironment {
+ // Callable that is responsible for environment setup.
+ optional ProcessCallable installation = 1;
+ // Callable, which is called to verify that an environment
+ // is suitable to start the UDF worker. This callable
+ // needs to verify that
+ // - The worker code itself is present
+ // - Any needed dependencies are present
+ optional ProcessCallable environment_verification = 2;
+ // Callable, which is invoked after the worker has been terminated.
+ // This can be used to cleanup dependencies or temporary resources.
+ // Be careful not to cleanup resources that could be used by
+ // other workers running in parallel.
+ optional ProcessCallable environment_cleanup = 3;
+}
+
+// Capabilities used for query planning
+message WorkerCapabilities {
+ // The data formats that the worker supports for UDF data in- & output.
+ // Every worker MUST at least support ARROW.
+ //
+ // It is expected that for each UDF execution, the input format
+ // always matches the output format.
+ //
+ // If a worker supports multiple data formats, the engine will select
+ // the most suitable one for each UDF invocation. Which format was chosen
+ // is reported by the engine as part of the UDF protocol's init message.
+ repeated UDFWorkerDataFormat supported_data_formats = 1;
+
+ // Which types of UDFs this worker supports.
+ // This should list all supported Shapes.
+ // Of which shape a specific UDF is will be communicated
+ // in the initial message of the UDF protocol.
+ //
+ // If a execution for an unsupported UDF type is requested
+ // the query will fail during query planning.
+ repeated UDFShape supported_udf_shapes = 2;
+
+ // Whether multiple, concurrent UDF
+ // connections are supported by this worker
+ // (for example via multi-threading).
+ //
+ // In the first implementation of the engine-side
+ // worker specification, this property will not be used.
+ //
+ // Usage of this property, can be enabled in the future if the
+ // engine implements more advanced resource management (Tbd).
Review Comment:
Erroneous comma separating subject from verb, and "Tbd" → "TBD":
```suggestion
// Usage of this property can be enabled in the future if the
// engine implements more advanced resource management (TBD).
```
##########
udf/worker/proto/src/main/protobuf/worker_spec.proto:
##########
@@ -17,12 +17,186 @@
syntax = "proto3";
+import "common.proto";
+
package org.apache.spark.udf.worker;
option java_package = "org.apache.spark.udf.worker";
option java_multiple_files = true;
-// Placeholder -- full worker specification schema to be added.
-// See design doc: SPIP SPARK-55278.
+///
+/// Worker specification
+///
message UDFWorkerSpecification {
+ WorkerEnvironment environment = 1;
+ WorkerCapabilities capabilities = 2;
+
+ // How to create new workers.
+ // At the moment, only direct creation is supported.
+ // This can be extended with indirect/provisioned creation in the future.
+ oneof worker {
+ DirectWorker direct = 3;
+ }
+}
+
+message WorkerEnvironment {
+ // Callable that is responsible for environment setup.
+ optional ProcessCallable installation = 1;
+ // Callable, which is called to verify that an environment
+ // is suitable to start the UDF worker. This callable
+ // needs to verify that
+ // - The worker code itself is present
+ // - Any needed dependencies are present
+ optional ProcessCallable environment_verification = 2;
+ // Callable, which is invoked after the worker has been terminated.
+ // This can be used to cleanup dependencies or temporary resources.
+ // Be careful not to cleanup resources that could be used by
+ // other workers running in parallel.
+ optional ProcessCallable environment_cleanup = 3;
+}
+
+// Capabilities used for query planning
+message WorkerCapabilities {
+ // The data formats that the worker supports for UDF data in- & output.
+ // Every worker MUST at least support ARROW.
+ //
+ // It is expected that for each UDF execution, the input format
+ // always matches the output format.
+ //
+ // If a worker supports multiple data formats, the engine will select
+ // the most suitable one for each UDF invocation. Which format was chosen
+ // is reported by the engine as part of the UDF protocol's init message.
+ repeated UDFWorkerDataFormat supported_data_formats = 1;
+
+ // Which types of UDFs this worker supports.
+ // This should list all supported Shapes.
+ // Of which shape a specific UDF is will be communicated
+ // in the initial message of the UDF protocol.
+ //
+ // If a execution for an unsupported UDF type is requested
+ // the query will fail during query planning.
+ repeated UDFShape supported_udf_shapes = 2;
+
+ // Whether multiple, concurrent UDF
+ // connections are supported by this worker
+ // (for example via multi-threading).
+ //
+ // In the first implementation of the engine-side
+ // worker specification, this property will not be used.
+ //
+ // Usage of this property, can be enabled in the future if the
+ // engine implements more advanced resource management (Tbd).
+ //
+ bool supports_concurrent_udfs = 3;
+
+ // Whether compatible workers may be reused.
+ // If this is not supported, the worker is
+ // terminated after every single UDF invocation.
+ bool supports_reuse = 4;
+
+ // To be extended with UDF chaining, ...
+}
+
+// The worker that will be created to process UDFs
+message DirectWorker {
+ // Blocking callable that is terminated by SIGTERM/SIGKILL
+ ProcessCallable runner = 1;
+
+ UDFWorkerProperties properties = 2;
+}
+
+message UDFWorkerProperties {
+ // Maximum amount of time to wait until the worker can accept connections.
+ //
+ // The engine will use this timeout, if it does not exceed a
+ // engine-configurable maximum time (e.g. 30 seconds).
+ optional int32 initialization_timeout_ms = 1;
+
+ // Used for graceful engine-initiated termination signaled via SIGTERM.
+ // After this time, the worker process should have terminated itself.
+ // Otherwise, the process will be forcefully killed using SIGKILL.
+ //
+ // The engine will use this timeout, if it does not exceed a
Review Comment:
```suggestion
// The engine will use this timeout, if it does not exceed an
```
##########
udf/worker/proto/src/main/protobuf/common.proto:
##########
@@ -0,0 +1,51 @@
+/*
+ * 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 org.apache.spark.udf.worker;
+
+option java_package = "org.apache.spark.udf.worker";
+
+option java_multiple_files = true;
+
+// The UDF in & output data format.
+enum UDFWorkerDataFormat {
+ UDF_WORKER_DATA_FORMAT_UNSPECIFIED = 0;
+
+ // The worker accepts and produces Apache arrow batches.
+ ARROW = 1;
+}
+
+// The UDF execution type/shape.
+message UDFShape {
+ oneof shape {
+ SparkUDFShapes spark = 1;
+ }
+}
+
+enum SparkUDFShapes {
Review Comment:
`SparkUDFShapes` uses plural naming, while `UDFWorkerDataFormat` in the same file uses singular. Proto convention recommends singular enum names — consider `SparkUDFShape`.
--
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
Re: [PR] [SPARK-56284] Adding UDF worker specification protobuf definition [spark]
Posted by "sven-weber-db (via GitHub)" <gi...@apache.org>.
sven-weber-db commented on code in PR #55165:
URL: https://github.com/apache/spark/pull/55165#discussion_r3072624778
##########
udf/worker/proto/src/main/protobuf/worker_spec.proto:
##########
@@ -17,12 +17,194 @@
syntax = "proto3";
+import "common.proto";
+
package org.apache.spark.udf.worker;
option java_package = "org.apache.spark.udf.worker";
option java_multiple_files = true;
-// Placeholder -- full worker specification schema to be added.
-// See design doc: SPIP SPARK-55278.
+///
+/// Worker specification
+///
message UDFWorkerSpecification {
+ WorkerEnvironment environment = 1;
+ WorkerCapabilities capabilities = 2;
+
+ // How to create new workers.
+ // At the moment, only direct creation is supported.
+ // This can be extended with indirect/provisioned creation in the future.
+ oneof worker {
+ DirectWorker direct = 3;
+ }
+}
+
+message WorkerEnvironment {
+ // Callable that is responsible for environment setup.
+ optional ProcessCallable installation = 1;
+ // Callable, which is called to verify that an environment
+ // is suitable to start the UDF worker. This callable
+ // needs to verify that
+ // - The worker code itself is present
+ // - Any needed dependencies are present
+ optional ProcessCallable environment_verification = 2;
+ // Callable, which is invoked after the worker has been terminated.
+ // This can be used to cleanup dependencies or temporary resources.
+ // Be careful not to cleanup resources that could be used by
+ // other workers running in parallel.
+ optional ProcessCallable environment_cleanup = 3;
+}
+
+// Capabilities used for query planning
+message WorkerCapabilities {
+ // The data formats that the worker supports for UDF data in- & output.
+ // Every worker MUST at least support ARROW.
+ //
+ // It is expected that for each UDF execution, the input format
+ // always matches the output format.
+ //
+ // If a worker supports multiple data formats, the engine will select
+ // the most suitable one for each UDF invocation. Which format was chosen
+ // is reported by the engine as part of the UDF protocol's init message.
+ repeated UDFWorkerDataFormat supported_data_formats = 1;
+
+ // Which types of UDFs this worker supports.
+ // This should list all supported Shapes.
+ // The shape of a specific UDF will be communicated
+ // in the initial message of the UDF protocol.
+ //
+ // If an execution for an unsupported UDF type is requested
+ // the query will fail during query planning.
+ repeated UDFShape supported_udf_shapes = 2;
+
+ // Whether multiple, concurrent UDF
+ // connections are supported by this worker
+ // (for example via multi-threading).
+ //
+ // In the first implementation of the engine-side
+ // worker specification, this property will not be used.
+ //
+ // Usage of this property can be enabled in the future if the
+ // engine implements more advanced resource management (TBD).
+ //
+ bool supports_concurrent_udfs = 3;
+
+ // Whether compatible workers may be reused.
+ // If this is not supported, the worker is
+ // terminated after every single UDF invocation.
+ bool supports_reuse = 4;
+
+ // To be extended with UDF chaining, ...
+}
+
+// The worker that will be created to process UDFs
+message DirectWorker {
+ // Blocking callable that is terminated by SIGTERM/SIGKILL
+ ProcessCallable runner = 1;
+
+ UDFWorkerProperties properties = 2;
+}
+
+message UDFWorkerProperties {
+ // Maximum amount of time to wait until the worker can accept connections.
+ //
+ // The engine will use this timeout, if it does not exceed an
+ // engine-configurable maximum time (e.g. 30 seconds).
+ optional int32 initialization_timeout_ms = 1;
+
+ // Used for graceful engine-initiated termination signaled via SIGTERM.
+ // After this time, the worker process should have terminated itself.
+ // Otherwise, the process will be forcefully killed using SIGKILL.
+ //
+ // The engine will use this timeout, if it does not exceed an
+ // engine-configurable maximum time (e.g. 30 seconds).
+ optional int32 graceful_termination_timeout_ms = 2;
+
+ // The connection this [[DirectWorker]] supports. Note that a single
+ // connection is sufficient to run multiple UDFs and (gRPC) services.
+ //
+ // On [[DirectWorker]] creation, connection information
+ // is passed to the callable as a string parameter.
+ // The string format depends on the [[WorkerConnection]]:
+ //
+ // For example, when using TCP, the callable argument will be:
+ // --connection PORT
+ // Here is a concrete example
+ // --connection 8080
+ //
+ // For the format of each specific transport type, see the comments below.
+ WorkerConnection connection = 3;
+}
+
+message WorkerConnection {
+ oneof transport {
+ UnixDomainSocket unix_domain_socket = 1;
+ TcpConnection tcp = 2;
+ }
}
+
+// Communication between the engine and worker
+// is done using a UNIX domain socket.
+//
+// On [[DirectWorker]] creation, a path to a socket
+// to listen on is passed as an argument.
+// Examples:
+// /tmp/channel-uuid.sock
+// /some/system/path/channel-1234.sock
+message UnixDomainSocket {}
+
+// Communication between the engine and worker
+// is done using a localhost TCP connection.
+//
+// On [[DirectWorker]] creation, a PORT
+// is passed as a connection parameter.
+//
+// It is expected that the worker binds to this
+// port on both IPv4 and IPv6 localhost interfaces.
+// E.g. the worker server should be reachable via
+// 127.0.0.1:PORT and [::1]:PORT.
+//
+// Examples:
+// 8080
+// 1234
+message TcpConnection {}
Review Comment:
Yes, good point. I renamed it `LocalTcpConnection`.
As for the port: This is the same as for the UDS. The port is engine-controlled and passed as a `--connection` flag to the UDF worker callable. This `LocalTcpConnection` message is there to capture additional TCP properties that the client might control in the future.
--
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
Re: [PR] [SPARK-56284] Adding UDF worker specification protobuf definition [spark]
Posted by "sven-weber-db (via GitHub)" <gi...@apache.org>.
sven-weber-db commented on code in PR #55165:
URL: https://github.com/apache/spark/pull/55165#discussion_r3072657808
##########
udf/worker/proto/src/main/protobuf/worker_spec.proto:
##########
@@ -17,12 +17,194 @@
syntax = "proto3";
+import "common.proto";
+
package org.apache.spark.udf.worker;
option java_package = "org.apache.spark.udf.worker";
option java_multiple_files = true;
-// Placeholder -- full worker specification schema to be added.
-// See design doc: SPIP SPARK-55278.
+///
+/// Worker specification
+///
message UDFWorkerSpecification {
+ WorkerEnvironment environment = 1;
+ WorkerCapabilities capabilities = 2;
+
+ // How to create new workers.
+ // At the moment, only direct creation is supported.
+ // This can be extended with indirect/provisioned creation in the future.
+ oneof worker {
+ DirectWorker direct = 3;
+ }
+}
+
+message WorkerEnvironment {
+ // Callable that is responsible for environment setup.
+ optional ProcessCallable installation = 1;
+ // Callable, which is called to verify that an environment
+ // is suitable to start the UDF worker. This callable
+ // needs to verify that
+ // - The worker code itself is present
+ // - Any needed dependencies are present
+ optional ProcessCallable environment_verification = 2;
+ // Callable, which is invoked after the worker has been terminated.
+ // This can be used to cleanup dependencies or temporary resources.
+ // Be careful not to cleanup resources that could be used by
+ // other workers running in parallel.
+ optional ProcessCallable environment_cleanup = 3;
+}
+
+// Capabilities used for query planning
+message WorkerCapabilities {
Review Comment:
Do you have a concrete name in mind?
For me, `Capabilities` already captures the notion of this message pretty well. This message contains everything the engine needs to plan and execute a UDF using this worker. E.g., it describes what the UDF worker is capable of (which data formats are supported, whether it supports concurrency, etc.).
I did update the comment:
```
// Capabilities used for query planning
```
to
```
// Capabilities used for query planning
// and running the worker during query execution.
```
To reflect that these properties are not only for query planning but also for execution
--
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
Re: [PR] [SPARK-56284] Adding UDF worker specification protobuf definition [spark]
Posted by "hvanhovell (via GitHub)" <gi...@apache.org>.
hvanhovell commented on code in PR #55165:
URL: https://github.com/apache/spark/pull/55165#discussion_r3073894594
##########
udf/worker/proto/src/main/protobuf/worker_spec.proto:
##########
@@ -17,12 +17,272 @@
syntax = "proto3";
+import "common.proto";
+
package org.apache.spark.udf.worker;
option java_package = "org.apache.spark.udf.worker";
option java_multiple_files = true;
-// Placeholder -- full worker specification schema to be added.
-// See design doc: SPIP SPARK-55278.
+///
+/// Worker specification
+///
message UDFWorkerSpecification {
+ // (Required)
+ WorkerEnvironment environment = 1;
+ // (Required)
+ WorkerCapabilities capabilities = 2;
+
+ // How to create new workers.
+ // At the moment, only direct creation is supported.
+ // This can be extended with indirect/provisioned creation in the future.
+ //
+ // (Required)
+ oneof worker {
+ DirectWorker direct = 3;
+ }
+}
+
+// Set of callables that can be used to setup, verify,
+// and cleanup the worker environment before the worker
+// callable is invoked.
+//
+// These callables should be specified if the worker
+// setup cannot be assumed to be pre-installed on the
+// target cluster.
+//
+// If defined, the below callables will be invoked once
+// per Spark executor in the following order (top -> bottom):
+//
+// ┌──────────────┐
+// │ │
+// │ Verification │
+// │ │
+// └──┬────────┬──┘
+// │ │
+// Failed │ │ Succeeded
+// │ │
+// ┌──────▼───────┐│
+// │ ││
+// │ Installation ││
+// │ ││
+// └──────┬───────┘│
+// │ │
+// ▼ ▼
+// ...
+// UDF worker creation
+// ...
+// │
+// │
+// ┌────▼────┐
+// │ │
+// │ Cleanup │
+// │ │
+// └─────────┘
+//
+// All scripts are optional.
+// However, if a verification script is supplied, an installation
+// script needs to be supplied as well.
+//
+message WorkerEnvironment {
+ // A callable that prepares cluster nodes by installing
+ // the required runtime, dependencies, and UDF worker binaries
+ // at the expected locations. This step ensures that
+ // the worker starting callable
+ // can subsequently locate and launch the worker process.
+ //
+ // (Required if environment_verification is given)
+ optional ProcessCallable installation = 1;
+
+ // Callable, which is called to verify that an environment
+ // is suitable to start the UDF worker. This callable
+ // needs to verify that
+ // - The worker code itself is present
+ // - Any needed dependencies are present
+ //
+ // (Optional)
+ optional ProcessCallable environment_verification = 2;
Review Comment:
Document that verification failure means that we need installation. Should be very specific about the error codes here, so we can also add support a fatal conditions as well (e.g. no-GPU, incompatible CPU arch, ...)?
--
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