zozo123 opened a new pull request, #58859:
URL: https://github.com/apache/spark/pull/58859
<!-- Draft: opened for design feedback on how the dispatcher factory should
be selected.
Happy to file a dedicated subtask under SPARK-55278 if that is the
preferred shape. -->
### What changes were proposed in this pull request?
Resolves the `TODO [SPARK-55278]: Wire in the correct dispatcher factory` in
`SparkEnv` by
selecting the `UDFDispatcherFactory` from a new configuration,
`spark.udf.worker.dispatcherFactory`, instead of installing a factory that
always throws.
### Why are the changes needed?
`SparkEnv.createUDFDispatcherManager` currently installs a
`UDFDispatcherFactory` whose
`createDispatcher` throws unconditionally, so no external UDF worker can be
created. I could
not find a subtask under SPARK-55278 covering this slot.
It cannot simply name a concrete dispatcher. `core` depends on
`udf-worker-proto` and
`udf-worker-core`; nothing in the reactor depends on `udf-worker-grpc`,
which owns
`DirectGrpcDispatcher` -- not `core`, `sql/core`, `assembly`, or the Connect
server.
`udf/worker/README.md` states this is deliberate:
> Keeping gRPC here means `proto/`, `core/`, and their consumers (`core`,
`catalyst`,
> `sql/core`) carry no gRPC dependency on their classpath.
So there is no module where the wiring code and a concrete dispatcher share
a classpath.
Closing the TODO by referencing a dispatcher at compile time would hand the
engine back the
gRPC dependency the module layout was built to avoid, and relocating the
wiring hits the same
wall. Resolving the factory reflectively preserves that boundary.
It also makes the extensibility the framework already documents reachable
without a fork:
`WorkerDispatcher` is described as decoupled from the provisioning model
("direct local spawn,
indirect daemon-provided"), and the README reserves `core/indirect/` for
"obtaining workers
from a provisioning service or daemon".
### Does this PR introduce _any_ user-facing change?
No behaviour change by default. When the config is unset the resolved
factory still throws,
and still only on first use, so applications that never execute an external
UDF are
unaffected. The message now names the config key rather than pointing at the
SPIP.
Adds `spark.udf.worker.dispatcherFactory` (optional, no default).
Deliberately not documented
in `configuration.md`: the surrounding APIs are `@Experimental` and not yet
wired to a
production execution path.
### How was this patch tested?
New `SparkEnvUDFDispatcherSuite` in `core`:
| Case | Expectation |
|---|---|
| config unset | resolution succeeds; `createDispatcher` throws naming the
config key |
| no-arg factory | instantiated and returned |
| `(SparkConf, isDriver)` factory | receives the conf and the correct
`isDriver` for both values |
| class not implementing `UDFDispatcherFactory` | `SparkException` naming
the config key and the class |
| unknown class name | `ClassNotFoundException` |
### Open questions
1. Is configuration-driven selection the intended resolution, or is
something else planned
(`ServiceLoader`, or giving a specific module the gRPC dependency)?
2. Config in `core` alongside `Python.scala`, or in the SQL namespace given
that external UDFs
are planned at the SQL layer?
3. `Utils.instantiateSerializerOrShuffleManager` does the right thing here
but is named for its
original callers -- reuse, or extract a generic helper?
--
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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
