ericm-db opened a new pull request, #57102:
URL: https://github.com/apache/spark/pull/57102
### What changes were proposed in this pull request?
Adds a second opt-in fast path for local Spark Connect development,
complementing the
persistent-server reuse mode of #56907 (SPARK-57787): a pool of
**pre-warmed, single-use** local
Connect servers.
With `SPARK_LOCAL_CONNECT_POOL=1` (or `.config("spark.local.connect.pool",
"true")`),
`SparkSession.builder.remote("local[*]").getOrCreate()`:
1. **claims** a fresh, never-used server from the pool directory
(`~/.spark/connect-local-pool`,
override via `SPARK_LOCAL_CONNECT_POOL_DIR`) -- a claim is an atomic
rename of the server's
discovery file, so exactly one racing client wins;
2. asynchronously spawns replacements to keep
`spark.local.connect.pool.size` (default 2) members
warm;
3. tears the claimed server down asynchronously when the session stops or
the client exits --
**no server ever serves two runs**.
Mechanics:
- Every member binds an ephemeral port with its own auth token, is seeded
with the spawning
client's start-up confs, and is stamped with a conf fingerprint; a run
only claims members whose
fingerprint matches its own confs, so a warm JVM is never handed to a run
whose static confs
differ from what it was booted with.
- A best-effort `fcntl` lock serializes spawn *accounting* so concurrent
cold starters share one
complement of launches instead of each spawning their own; claims
themselves are lock-free
atomic renames. Platforms without `fcntl` may transiently over-spawn;
extras are reclaimed by
the idle timeout (same degradation posture as the reuse path's start lock).
- A janitor reaps stale state on every acquire: launches whose daemon died,
unreachable or
version-mismatched members, servers claimed by clients that died uncleanly
(e.g. SIGKILL), and
retired servers that hang in shutdown (SIGTERM at release, escalated to a
hard kill after 30s).
- Server-side backstops: pool members run with `--exit-after-use`
(self-terminate once their one
client's session has come and gone) and an idle timeout
(`SPARK_LOCAL_CONNECT_POOL_IDLE_TIMEOUT`, default 1800s), so an idle
machine drains to zero.
- `SparkSession._purge_local_connect_pool()` force-kills every member (warm,
in-flight, or
claimed) and empties the pool directory -- a one-command "really clean
state" escape hatch.
Relationship to the reuse mode (#56907): reuse keeps one long-lived server
that all runs
reconnect to -- cheapest, but state backed by the shared `SparkContext`
(persistent catalog,
global temp views, cached data) carries across runs and the JVM ages. The
pool trades standing
memory (~550 MB RSS per warm member measured on macOS with default confs;
the default pool of 2
is about 1.1 GB while actively iterating) for per-run isolation identical to
today's default
behavior. This directly addresses the determinism concern raised on the dev
list about reusing a
long-lived backend server:
https://lists.apache.org/thread/sg9o2gbb3nttz74f0s01v8f167zy8ltt
**Stacked on #56907**: the earlier commits belong to that PR; only the last
commit is new here.
Will rebase once #56907 merges.
### Why are the changes needed?
Local `getOrCreate()` pays a multi-second JVM + `SparkContext` cold start in
every process.
#56907 removes that cost by reusing one long-lived server, but reuse was
flagged on the dev list
as a determinism/isolation risk if it ever became a default: a shared
backend can be left in a
bad state that no fresh environment reproduces. The pool keeps the latency
win (measured on a dev
laptop: ~0.35-0.4s to a usable session vs ~3.2-4.6s cold) while preserving
fresh-server-per-run semantics, making it a candidate for an eventual
default-on story.
### Does this PR introduce _any_ user-facing change?
Only when the opt-in is enabled. With `SPARK_LOCAL_CONNECT_POOL=1` (or
`spark.local.connect.pool=true`), a `local` remote `getOrCreate()` claims a
pre-warmed, never-used
local Connect server and replaces it in the background, instead of booting
an in-process server.
With the opt-in unset (the default), behavior is unchanged. If both this and
`spark.local.connect.reuse` are set, the pool takes precedence.
### How was this patch tested?
New `python/pyspark/sql/tests/connect/test_connect_local_server_pool.py`:
- Unit tests (no real servers): claim atomicity, fingerprint matching, and
consumption of launch
bookkeeping; janitor rules (dead launches, unreachable members, claims
orphaned by SIGKILLed
clients, retired-server escalation, conf-seed cleanup) with live process
groups standing in for
servers; release writes the retired marker and stops the server; purge
kills all members and
empties the directory; pool lock roundtrip; pool-size parsing.
- End-to-end (real servers): sequential runs get distinct fresh servers and
each used server is
torn down afterwards; two concurrent cold clients get distinct servers.
Also verified manually on macOS: two simultaneous cold clients share one
spawn complement under
the pool lock; a global temp view and a cached table created in one run are
not visible to the
next (the state that leaks by design under the reuse mode); a SIGKILLed
client's orphaned server
is reaped by the next run's janitor while that run still connects in ~0.4s;
purge leaves zero
pool processes; idle RSS per warm member measured at 534-557 MB.
### Was this patch authored or co-authored using generative AI tooling?
Generated-by: Claude Code (Fable 5)
--
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]
