[SPARK-57787][CONNECT] Reuse a persistent local Spark Connect server managed by sbin/start-connect-server.sh#57147
Draft
ericm-db wants to merge 14 commits into
Draft
[SPARK-57787][CONNECT] Reuse a persistent local Spark Connect server managed by sbin/start-connect-server.sh#57147ericm-db wants to merge 14 commits into
ericm-db wants to merge 14 commits into
Conversation
…for faster local startup
Adds an opt-in (SPARK_LOCAL_CONNECT_REUSE / spark.local.connect.reuse) so that
`SparkSession.builder.remote("local[*]").getOrCreate()` reconnects to a persistent local
Spark Connect server -- starting a detached one on the first run and reconnecting to it on
later runs -- instead of booting a fresh in-process server in every process. The first run
pays the cold start once; later runs reconnect in a fraction of a second.
Default behavior is unchanged when the opt-in is off. No protocol or Scala changes.
Trim verbose comments/docstrings, drop redundant noqa and the private-method example from the user docs, and consolidate exception handling. No functional change.
…d seed server confs Fixes two issues raised in review of the persistent local Connect server path: 1. Concurrent first-time startup could fail spuriously. Opted-in processes starting at the same time (or after a version upgrade) would each spawn a server on the fixed port; the loser's JVM failed to bind and raised LOCAL_CONNECT_SERVER_START_FAILED even though a usable server was now up. Serialize start-up with a discovery-file lock, reconnect to the winner if our own daemon exits, and fall back to an ephemeral port when the configured one is already held. 2. The daemon dropped most builder options on first startup. Forward the caller's start-up confs (warehouse dir, jars/packages, catalog, app name, etc.) to the server via a JSON conf file so first-run behavior matches the in-process path. A later run reconnecting to an already-warm JVM still cannot change its static confs, as documented.
On the start-up timeout path, terminating only the detached daemon process left its child JVM orphaned when the timeout fired before the daemon had installed its own signal handling (the slow- startup case that triggers the timeout). Signal the whole process group -- the daemon is a session leader and the JVM stays in its group -- and escalate to SIGKILL if a graceful stop does not take, so the JVM is reaped instead of leaking.
Adds a POSIX-only test that starts the detached daemon, waits until it has spawned its child JVM, then calls _terminate_local_connect_server and asserts the whole process group is gone -- covering the orphaned-JVM case the timeout-path fix addresses. Also waits for the server port to close in tearDown so a stopped server's JVM cannot linger into the next test.
…ing the local server The stop path signals the pid recorded in the discovery file via killpg. If that pid is stale (e.g. the daemon was SIGKILLed and left its discovery file behind) and has been recycled by an unrelated process, group-killing it could take down the caller's own process group. The daemon is always launched as a session leader, so its group id equals its pid; only signal the group when that holds, and fall back to a plain kill otherwise. Adds a unit test covering both directions (with the signal syscalls mocked so a regression cannot kill the test runner), and a ruff-format fix for a missing blank line after the local_server.py module docstring. Co-authored-by: Isaac
…ves the daemon test_terminate_reaps_daemon_and_jvm fails in CI with "ProcessLookupError not raised": _terminate_local_connect_server keyed SIGKILL escalation on the daemon timing out, so when the daemon exits promptly on SIGTERM while its JVM's graceful shutdown lingers (or wedges), nothing ever escalates and the group survives. Terminate now waits for the whole process group to disappear after the daemon settles and group-SIGKILLs any survivors past a 10s grace period. It signals the group id directly since the leader may already be reaped by then, and polls the daemon so a zombie leader does not keep the group looking alive. Adds a deterministic regression test (a session leader that exits on SIGTERM after spawning a SIGTERM-immune child standing in for the slow JVM), and hardens both terminate tests against macOS zombie semantics: killpg on a group holding only a zombie yields EPERM, and kill(pid, 0) succeeds for a zombie until it is reaped. Co-authored-by: Isaac
…te tests Both terminate tests fail in CI with "process group survived": the tests checked group death via killpg(pgid, 0), but a zombie keeps its group signalable, and in CI containers whose pid 1 never reaps, the daemon's JVM reparents on the daemon's death and lingers as a zombie indefinitely -- so the check never turns negative even though nothing is running. (macOS launchd reaps promptly, which is why the tests pass locally.) The tests now assert that no *running* (non-zombie) process remains in the group, using ps to inspect member states. Co-authored-by: Isaac
…tion Review feedback: prefer fenced code blocks over Liquid highlight tags in the new documentation section; they render the same and keep raw-Markdown editors from mistaking comment hashes for headings. Co-authored-by: Isaac
… run the daemon via -m Addresses review comments: - Launch the daemon with `python -m pyspark.sql.connect.local_server` instead of by file path, dropping the sys.path workaround; the daemon always runs from the client's environment, which has the Connect client dependencies. - Move all client-side reuse helpers off SparkSession (they were all staticmethods) into pyspark/sql/connect/local_server.py as module functions; connect/session.py is back to its upstream state. - Pass the auth token only through SPARK_CONNECT_AUTHENTICATE_TOKEN (same precedence as the in-process path); it no longer appears on the daemon argv (visible in ps) nor as a duplicate conf (the server falls back to the env var when the conf is unset). - Probe pid liveness only on POSIX: os.kill(pid, 0) terminates the target process on Windows. - Close the remove-vs-publish race on the discovery file by claiming the start-up lock non-blockingly in _remove_discovery_if_ours. - Turn the start-up lock into a contextmanager; validate discovery value types on read and index the dict directly afterwards; simplify _discovery_path and the makedirs guards; drop the SIGINT try/except (main only runs on the main thread); reap the SIGKILLed daemon with a short wait instead of a bare poll. - Lower the default idle timeout to 1800s, add --stop to the module CLI, and update the docs accordingly. Co-authored-by: Isaac
…rt-connect-server.sh Reworks the opt-in local Connect server reuse path per review feedback: instead of a detached Python daemon that boots a classic Py4J session with the Connect plugin and babysits it, the launcher now drives the standard `sbin/start-connect-server.sh` (spark-daemon.sh submit SparkConnectServer) and lets spark-daemon.sh own the JVM (pid file, logs, daemonization). - No Python server process any more: PySpark runs the script with the auth token in its env, waits for the port, writes the discovery file (pid taken from the spark-daemon pid file), and returns the endpoint. - Deletes the Py4J usage, signal handling, idle-session polling, process-group management and SIGKILL escalation, and the discovery-file removal race guard -- none of it has an equivalent in the new shape. - Idle self-termination goes away with the daemon; the server runs until stopped (`python -m pyspark.sql.connect.local_server --stop`, or `sbin/stop-connect-server.sh`). `spark.local.connect.server.idleTimeout` is removed. - The server port is picked client-side (configured port when free, else an OS-assigned one), since the standalone script cannot report a bound ephemeral port back. - Seed confs travel via a 0600 spark-submit --properties-file instead of a JSON file parsed by the daemon. Artifact isolation needs no special handling: SparkConnectServer enables it itself. - The reuse path is POSIX-only now (it drives the sbin shell scripts) and says so explicitly; server logs land next to the discovery file (~/.spark/logs by default) for debuggability. - Ships start/stop-connect-server.sh in the pip sbin package (their dependencies, bin/* and spark-daemon.sh, were already packaged). Net effect vs the previous approach: -313 lines, and the server lifecycle is exactly the one users get from the documented manual workflow. Co-authored-by: Isaac
…e manual server workflow
Per review discussion, spark.local.connect.reuse / SPARK_LOCAL_CONNECT_REUSE
stays an internal, undocumented conf. The docs now cover only the explicit
workflow (sbin/start-connect-server.sh + .remote("sc://...")) plus a note on
what is and is not isolated between runs against a shared local server.
Co-authored-by: Isaac
Contributor
Author
|
@HyukjinKwon Is this more along the lines of what you were thinking? |
Co-authored-by: Isaac
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
1 participant
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
What changes were proposed in this pull request?
Opt-in reuse of a persistent local Spark Connect server for fast local iteration, implemented by driving the standard
sbin/start-connect-server.shscript.This is the successor of #56907, restructured per #56907 (comment): instead of a detached Python daemon that boots a classic Py4J session with the Connect plugin and babysits it, PySpark now starts the server through
sbin/start-connect-server.sh(spark-daemon.sh submit SparkConnectServer) and letsspark-daemon.shown the process (daemonization, pid file, logs). The branch stacks on #56907, so the incremental diff between the two approaches is visible here: ericm-db/spark@local-connect-reuse...local-connect-reuse-sbin (net -313 lines).Mechanics, all in
python/pyspark/sql/connect/local_server.py:spark.local.connect.reuse/SPARK_LOCAL_CONNECT_REUSEis set and the master islocal[...],SparkSession.buildercallsreuse_or_start_local_connect_server()instead of booting a fresh in-process server.sbin/start-connect-server.shwith the auth token in its environment (never argv), waits for the port, and writes a0600discovery file (host, port, token, pid from the spark-daemon pid file, Spark version) under~/.spark/(relocatable viaSPARK_LOCAL_CONNECT_DISCOVERY).SparkConnectServerenables artifact isolation itself).python -m pyspark.sql.connect.local_server --stoporsbin/stop-connect-server.sh. Server logs land next to the discovery file (~/.spark/logs).0600spark-submit--properties-file.start-connect-server.sh/stop-connect-server.sh(their dependencies,bin/*andspark-daemon.sh, were already packaged).sbin/scripts and raises a clear error on Windows.Why are the changes needed?
SparkSession.builder.remote("local[*]").getOrCreate()boots a new JVM and Connect server in every Python process, so eachpython script.pyiteration re-pays several seconds of cold start. Reconnecting to one long-lived local server makes the edit/run loop sub-second. This is off by default and adds no new server mechanism: it automates exactly the documented manual workflow (start-connect-server.sh+.remote("sc://localhost:15002")).Does this PR introduce any user-facing change?
No change unless explicitly opted in. When opted in (
spark.local.connect.reuseconf /SPARK_LOCAL_CONNECT_REUSEenv var, plus optionalspark.local.connect.server.port),.remote("local[*]")reconnects to a persistent local server instead of booting an in-process one. Newpython -m pyspark.sql.connect.local_server --stopcommand. pip installs additionally ship the connect start/stop sbin scripts. Per review discussion the opt-in conf is intentionally internal and undocumented; the docs change only describes the existing explicit workflow (sbin/start-connect-server.sh+.remote("sc://...")) and what is / is not isolated between runs sharing a local server.How was this patch tested?
New test suite
pyspark.sql.tests.connect.test_connect_local_server(19 tests, wired into thepyspark_connectmodule): discovery-file validation (missing/malformed/typed fields), reuse decision (version mismatch, dead pid, closed port, live server; the pid probe is asserted to never run on Windows, whereos.kill(pid, 0)terminates the target), port picking, seed properties file permissions/format, start-lock round trip, stop with and without a running server, the--stopCLI, and POSIX gating. End-to-end tests start realspark-daemon.sh-managed servers: builder opt-in, three concurrent cold-starting clients converging on a single server, reconnect + server-side session isolation, and static conf seeding.Was this patch authored or co-authored using generative AI tooling?
Generated-by: Claude Code (claude-fable-5)
This pull request and its description were written by Isaac.