All posts

Spark Connect: the driver becomes a server, and your application becomes a client

Spark Connect splits a Spark application in two: a thin client that builds unresolved plans, and a server that analyses and runs them. Here is how the protocol works, when each piece arrived, what it buys and what it costs, and every behaviour below run from a client container with no Java and no Spark installed.

32 min read Spark

TL;DR

  • Spark Connect turns the driver into a server and your program into a client that sends unresolved logical plans over gRPC and receives results as Arrow batches. The client needs no JVM: the Python client is a 1.6 MB package, against 455 MB for full PySpark.
  • Analysis happens on the server, and later. A misspelt column is not an error when you define the DataFrame, only when something first needs its schema.
  • Sessions are isolated, the server is shared. Each client session has its own temporary views and SQL settings; tables, executors and the cluster are common to everyone connected.
  • Your Python code still runs on the server. UDFs execute in Python workers next to the server, so the client and server must use the same Python minor version, and modules your UDFs import have to be shipped with addArtifacts.
  • The server has no authentication by default. Anyone who can reach its port can run code on your cluster. Put a token and TLS, or an authenticating proxy, in front of it before exposing it.
  • What does not exist on the client is anything that needs the JVM: sparkContext, RDDs, _jdf, broadcast variables and accumulators.

A classic PySpark program runs a Spark driver inside your own process: import pyspark starts a Java virtual machine, and that JVM is the Spark driver: it holds the session, parses and optimises every query, schedules every task, and collects every result. Your notebook, your Airflow task and your web service each carry a whole Spark driver around with them, with its memory, its classpath and its version.

Spark Connect removes the driver from your process. What is left behind is a client that describes queries and a server that runs them, talking over a documented protocol. This post explains how that split works, when each part of it shipped, and what it changes in practice. Every behaviour quoted below was run for this post: a Spark 4.1.3 Connect server in one container, and clients in two plain python:3.10-slim and python:3.12-slim containers that had only pip install pyspark-client and never had Java installed.

Why does Spark Connect exist?

In a classic Spark application the driver runs inside the client process. That design made Spark easy to start with and hard to operate at scale, for five reasons that all come from the same coupling:

  • Every client is a full driver. A notebook user who runs collect() on a large table can exhaust the driver’s memory, and the driver is their process. On a shared cluster, a single user’s mistake can take the driver, and everyone’s work on it, down with it.
  • Dependencies collide. The client’s Python or Java dependencies share a classpath or an interpreter with Spark’s own. Upgrading a library in your application can break Spark, and upgrading Spark can break your application.
  • Upgrades are all-or-nothing. Because each client embeds a driver, moving the cluster to a new Spark version means moving every client to it at the same time.
  • Only JVM-adjacent languages are practical. Python works because PySpark drives a JVM through Py4J. A Go or Rust program would have to do the same, or go without Spark.
  • Thin environments cannot run Spark at all. An IDE plugin, a web application or a small container would need Java and hundreds of megabytes of JARs just to submit a query.

Spark Connect’s answer is to put a network boundary where the coupling was:

flowchart LR
  subgraph CL["Classic: one process"]
    direction TB
    A1["your code"] --> B1["Py4J"] --> C1["driver JVM<br/>session, analyzer, optimizer,<br/>scheduler, results"]
  end
  C1 --> E1["executors"]

  subgraph CN["Spark Connect: two processes"]
    direction TB
    A2["your code"] --> B2["thin client<br/>builds unresolved plans"]
    B2 -->|"gRPC<br/>protobuf plans"| C2["Connect server = driver<br/>session, analyzer, optimizer,<br/>scheduler"]
    C2 -->|"Arrow batches"| B2
  end
  C2 --> E2["executors"]

The driver still exists. It simply lives in a long-running server process that many clients share, instead of inside each of them.

Architecture: how a query travels

How it works. The client does not know what a table is, cannot resolve a column, and never optimises anything. When you write DataFrame code, each call only adds a node to an unresolved logical plan, a tree of relations and expressions that refers to tables and columns by name, encoded as a protocol buffer. Nothing leaves the client until an action or a schema lookup needs the server. Then one of a small set of gRPC calls carries the plan across:

  • ExecutePlan runs a plan and streams the results back as a sequence of Arrow record batches, plus metrics, as the server produces them.
  • AnalyzePlan asks for something about a plan without running it: its schema, its explain output, whether it is streaming. df.columns and df.schema use it.
  • Config reads and writes runtime SQL settings for the session.
  • AddArtifacts uploads files the server needs: Python modules, JARs, archives, and large local data.
  • Interrupt cancels running operations, by operation ID or by tag.
  • ReattachExecute and ReleaseExecute let a client reconnect to an operation that is still running after its network connection dropped, and tell the server when it has read everything.
sequenceDiagram
  participant C as client (no JVM)
  participant S as Connect server (driver)
  participant X as executors
  C->>C: df = spark.table("orders").where(...).select(...)<br/>builds unresolved plan, sends nothing
  C->>S: AnalyzePlan(plan)  [df.columns]
  S-->>C: schema
  C->>S: ExecutePlan(plan, operation_id, tags)  [df.collect()]
  S->>S: resolve names against the catalog,<br/>optimise, plan physically
  S->>X: tasks
  X-->>S: results
  S-->>C: stream of Arrow batches + metrics
  C->>S: ReleaseExecute(operation_id)

On the server, a session manager keeps one session per (user_id, session_id) pair. Each holds an ordinary Spark session, so everything after the plan arrives is the same Catalyst, the same AQE and the same executors as a classic application. The difference is entirely in who holds the session and how plans reach it.

Good to know: because the client sends plans rather than SQL strings or bytecode, any language that can build the protobuf messages and read Arrow can be a Spark client. That is why there are Python, Scala, Java, Go, Rust and Swift clients, and a JDBC driver, all speaking to the same server.

When did each piece arrive?

Spark Connect was proposed as a Spark Project Improvement Proposal in 2022 (SPARK-39375) and has been built out in every release since. The pattern is the usual one for Spark: introduced, made usable, then made the foundation other features are built on.

flowchart LR
  A["<b>3.4</b><br/>Apr 2023<br/>Python client,<br/>first Scala client"] --> B["<b>3.5</b><br/>Sep 2023<br/>Scala and Go clients,<br/>streaming, pandas API"]
  B --> C["<b>4.0</b><br/>May 2025<br/>pyspark-client,<br/>spark.api.mode, ML"]
  C --> D["<b>4.1</b><br/>Dec 2025<br/>JDBC driver,<br/>idempotent reattach"]
  D --> E["<b>4.2</b><br/>Jul 2026<br/>GetStatus API,<br/>history server tab"]

Spark 3.4, April 2023: the Python client

Tickets: SPARK-39375.

The first release. The Python client implemented the DataFrame, Column, Functions, SparkSession, I/O and Catalog APIs, Python UDFs, the pandas and Arrow function APIs (mapInPandas, applyInPandas), and runtime SQL configuration. A basic Scala client arrived at the same time, with a REPL. The server was started with start-connect-server.sh and needed the Connect JAR passed with --packages.

Spark 3.5, September 2023: more clients, and streaming

Tickets: SPARK-42554, SPARK-43351, SPARK-42938.

The Scala client became a real client, helped by splitting Spark’s SQL module into sql and a dependency-light sql-api that client and server share. An initial Go client appeared in its own repository. Structured Streaming worked from Python and Scala clients, the pandas API on Spark worked over Connect, and TorchDistributor brought distributed PyTorch training. Request handling improved too: asynchronous execution, retries, and long-lived queries that survive a dropped connection.

Spark 4.0, May 2025: Connect as a supported way to run Spark

Tickets: SPARK-50605, SPARK-50812, SPARK-49248.

  • pyspark-client, a pure-Python package with no JARs. The 4.1.3 source distribution I downloaded is 1.6 MB; the full pyspark one is 455 MB.
  • spark.api.mode, a setting that makes an ordinary PySpark program use Spark Connect without changing its code.
  • A release tarball with Spark Connect enabled by default.
  • API parity work for the Scala client, a unified Scala interface shared by classic and Connect, and parent classes that let code accept either kind of session.
  • ML on Spark Connect, so pyspark.ml works from a remote client.
  • A Swift client, alongside the Go and Rust clients in their own repositories.

Spark 4.1, December 2025: tools, and robustness

Tickets: SPARK-53484, SPARK-52397, SPARK-53455.

  • A JDBC driver that speaks the Connect protocol (jdbc:sc://host:15002), so JDBC tools reach a Connect server without the Thrift server.
  • Idempotent ExecutePlan: sending the same operation ID and plan a second time reattaches to the running operation instead of starting it again, which makes client retries safe.
  • A CloneSession RPC, gRPC status codes on Python exceptions, server-side column name validation, transformWithState over Connect, and ML on Connect declared generally available for Python.
  • New components built on Connect: the spark-pipelines CLI for Declarative Pipelines is a Connect client.

Spark 4.2, July 2026: observability and RDD-style gaps

Tickets: SPARK-55606, SPARK-57601, SPARK-55227.

  • A GetStatus API for monitoring an operation’s execution status.
  • The Spark Connect tab in the history server, so finished sessions can be inspected after the server restarts.
  • head(), take() and tail() optimised to avoid full scans.
  • APIs that fill gaps people used RDDs for: zipWithIndex on DataFrames and Datasets, DataFrame.toJSON in Python, and spark.read.json, csv and xml accepting a DataFrame of strings.
  • A client-side limit on the size of local data, and an option to release the remote session when the client process exits (SPARK_CONNECT_RELEASE_SESSION_ON_EXIT).

Running a server and a thin client

Start the server

Every Spark 4.x distribution includes the server. It is an ordinary Spark application, so it takes the usual spark-submit options: master, memory, packages for connectors, and configuration:

$SPARK_HOME/sbin/start-connect-server.sh \
  --master "local[4]" \
  --driver-memory 4g \
  --conf spark.connect.grpc.binding.port=15002 \
  --conf spark.sql.warehouse.dir=/tmp/wh

It runs as a daemon and logs to $SPARK_HOME/logs/; set SPARK_NO_DAEMONIZE=1 to keep it in the foreground, which is what you want in a container. On 3.4 and 3.5 the server also needs its JAR: --packages org.apache.spark:spark-connect_2.12:3.5.9.

Good to know: anything that must be installed into the session, such as a table format’s JARs, spark.sql.extensions and catalog settings, has to be on the server’s command line. A client can change runtime SQL settings, but cannot load an extension into a session that already exists. The Iceberg and Hudi posts show complete server command lines for each format.

Install the client

pip install pyspark-client==4.1.3      # 1.6 MB; pulls grpcio, pandas and pyarrow

Pin it to the server’s version: an unpinned pip install fetches the newest client on PyPI, whatever your server runs. A client newer than its server can half work, which is worse than failing. A pinned 4.1.3 client against a 3.5.9 Connect server:

import pyspark
from pyspark.sql import SparkSession

spark = SparkSession.builder.remote("sc://sc35-server:15002").getOrCreate()
print("client", pyspark.__version__, "-> server", spark.version)
print("range:", spark.range(3).count())
try:
    print("createDataFrame:", spark.createDataFrame([(1, "pune")], "id INT, city STRING").collect())
except Exception as e:
    print("createDataFrame failed:", str(e).splitlines()[0][:220])
spark.stop()
# client 4.1.3 -> server 3.5.9
# range: 3
# createDataFrame failed: [SQL_CONF_NOT_FOUND] The SQL config "spark.sql.session.localRelationChunkSizeRows" cannot be found. Please verify that the config exists.

spark.range only sends a plan, so it succeeded. createDataFrame from local data makes the 4.1 client ask the server for a setting that controls how local rows are chunked for upload, and 3.5.9 has never heard of it. SQL fails too, less helpfully: from the same 4.1.3 client, spark.sql(...) against the 3.5.9 server returned PARSE_SYNTAX_ERROR “at or near end of input”, because the statement arrives at the older server as an empty string. The reverse direction behaved: a pyspark[connect]==3.5.9 client against a 4.1.3 server ran the same script, createDataFrame included, and printed [Row(id=1, city='pune')].

Connect

The client connects with a URL instead of a master:

import os, sys, shutil
from pyspark.sql import SparkSession

spark = SparkSession.builder.remote("sc://sc-server:15002").getOrCreate()
print("client python:", sys.version.split()[0], "| java on client:", shutil.which("java"))
print("session class:", type(spark).__module__ + "." + type(spark).__name__)
print("server version:", spark.version)
print("session id looks like a UUID:", len(spark.client._session_id) == 36)
spark.range(5).selectExpr("id", "id * id AS sq").show()
# client python: 3.10.21 | java on client: None
# session class: pyspark.sql.connect.session.SparkSession
# server version: 4.1.3
# session id looks like a UUID: True
# +---+---+
# | id| sq|
# +---+---+
# |  0|  0|
# |  1|  1|
# |  2|  4|
# |  3|  9|
# |  4| 16|
# +---+---+
spark.stop()

No java on the client, a session class from pyspark.sql.connect, and a query that ran on a 4.1.3 server. The URL has the form sc://host:port/;param=value;param=value, and the parameters the client understands are token, use_ssl, user_id, user_agent and session_id. SPARK_REMOTE=sc://host:15002 in the environment does the same as .remote().

What travels over the wire

How it works. You can look at the plan the client is about to send. A range, a filter and a projection become a small protobuf message in which every column reference and function is still just a name:

from pyspark.sql import SparkSession
from pyspark.sql import functions as F

spark = SparkSession.builder.remote("sc://sc-server:15002").getOrCreate()
df = (spark.range(0, 1000)
        .where(F.col("id") % 7 == 0)
        .select((F.col("id") * 10).alias("x")))
proto = df._plan.to_proto(spark.client)
text = str(proto)
print(type(proto).__module__ + "." + type(proto).__name__, "|", len(proto.SerializeToString()), "bytes on the wire")
print("\n".join(text.splitlines()[:28]))
spark.stop()

pyspark.sql.connect.proto.base_pb2.Plan | 399 bytes on the wire
root {
  common {
    plan_id: 2
  }
  project {
    input {
      common {
        plan_id: 1
      }
      filter {
        input {
          common {
            plan_id: 0
          }
          range {
            start: 0
            end: 1000
            step: 1
          }
        }
        condition {
          unresolved_function {
            function_name: "=="
            arguments {
              unresolved_function {
                function_name: "%"
                arguments {
                  unresolved_attribute {

A few hundred bytes describe the whole query. The nesting is the plan tree read inside out: a project whose input is a filter whose input is a range. The condition is unresolved_function "==" over unresolved_function "%" over an unresolved_attribute: the client has no idea whether id exists or what type it has. The server will work that out.

The size is not fixed, and the reason is useful. Running the same code from a file with a different name produced 528 bytes instead of 399, because every expression carries a python_origin with its call site, such as call_site: "/w/post_blocks/proto_check.py:4". That is how an analysis error raised on the server can point at the line of your code that built the bad expression, even though the error happened in another process.

Good to know: _plan is an internal attribute, useful for learning and debugging, not an API to build on. df.explain() works over Connect too; it sends an AnalyzePlan and prints the plan the server produced.

Analysis happens on the server, and later

How it works. In a classic session, select("no_such_column") fails immediately, because the driver in your process resolves each DataFrame as you build it. Over Connect the client cannot resolve anything, so a DataFrame with a mistake in it is perfectly happy to exist:

from pyspark.sql import SparkSession
from pyspark.errors import PySparkException

spark = SparkSession.builder.remote("sc://sc-server:15002").getOrCreate()
df = spark.range(3).select("no_such_column")      # nothing is sent yet
print("DataFrame defined without error:", type(df).__module__)
try:
    df.columns                                     # needs the schema: an AnalyzePlan RPC
except PySparkException as e:
    print("error on first schema use:", e.getCondition(), e.getSqlState())
# DataFrame defined without error: pyspark.sql.connect.dataframe
# error on first schema use: UNRESOLVED_COLUMN.WITH_SUGGESTION 42703
spark.stop()

The error arrives with its error class and SQLSTATE intact, carried across gRPC, so error handling written against a classic session keeps working.

Good to know: two practical consequences. Errors surface further from the line that caused them, often at a collect() or write many lines later, so tests should trigger analysis early (df.schema is enough). And every df.columns or df.schema is a network round trip; code that inspects a schema inside a loop, for example to add hundreds of columns one at a time, becomes slow over Connect. Build the column list once and make a single select.

Sessions: what is isolated and what is shared

How it works. The server keeps one session per client session ID, and a single Python process can hold several with create() instead of getOrCreate():

from pyspark.sql import SparkSession

url = "sc://sc-server:15002"
a = SparkSession.builder.remote(url).create()
b = SparkSession.builder.remote(url).create()
print("different sessions:", a.client._session_id != b.client._session_id)

a.range(3).createOrReplaceTempView("only_in_a")
a.conf.set("spark.sql.shuffle.partitions", "7")
a.range(3).write.mode("overwrite").saveAsTable("shared_table")

print("a sees its view:", a.catalog.tableExists("only_in_a"))
print("b sees a's view:", b.catalog.tableExists("only_in_a"))
print("shuffle.partitions in a / b:", a.conf.get("spark.sql.shuffle.partitions"), "/", b.conf.get("spark.sql.shuffle.partitions"))
print("b sees the table:", b.table("shared_table").count(), "rows")
# different sessions: True
# a sees its view: True
# b sees a's view: False
# shuffle.partitions in a / b: 7 / 200
# b sees the table: 3 rows
a.sql("DROP TABLE shared_table")
a.stop(); b.stop()

Session b reports 200, the server’s default; when I re-ran the block against a server started with --conf spark.sql.shuffle.partitions=8, it reported 8. Temporary views, SQL settings, registered functions and the current database belong to a session. Tables in the catalog, cached data, the executors and the cluster’s capacity belong to the server, and every session shares them.

Good to know: the server removes sessions that have been idle for spark.connect.session.manager.defaultSessionTimeout, 60 minutes by default, along with their temporary views. A long-lived client that pauses for an hour comes back to a fresh session. And because executors are shared, one session’s heavy query slows everyone else’s; run the server with the fair scheduler and separate pools for interactive and batch users.

Python UDFs run on the server

How it works. A Python UDF defined on the client is pickled, with its closure, and sent inside the plan. The server starts Python workers next to its executors and runs the function there, exactly as a classic session would. So the function’s Python has to be compatible with the server’s. My first client container had Python 3.12 and the server image had 3.10:

import sys
from pyspark.sql import SparkSession
from pyspark.sql.functions import udf

spark = SparkSession.builder.remote("sc://sc-server:15002").getOrCreate()
print("client python", sys.version.split()[0])

@udf("string")
def shout(s):
    return s.upper() + "!"

spark.createDataFrame([("pune",)], "city STRING").select(shout("city").alias("c")).show()
spark.stop()

On the 3.12 client:

client python 3.12.14
pyspark.errors.exceptions.base.PySparkRuntimeError: [PYTHON_VERSION_MISMATCH] Python in worker has
different version: 3.10 than that in driver: 3.12, PySpark cannot run with different minor versions.

On the 3.10 client, the same script:

client python 3.10.21
+-----+
|    c|
+-----+
|PUNE!|
+-----+

Plain DataFrame and SQL code worked from both clients; only the UDF cares, because only the UDF is Python code that has to run on the other side.

The same reasoning applies to modules your UDF imports. They exist on the client, not on the server, until you upload them:

from pyspark.sql import SparkSession
from pyspark.sql.functions import udf

spark = SparkSession.builder.remote("sc://sc-server:15002").getOrCreate()

@udf("double")
def with_gst(amount):
    import mylib                     # a module that exists only on the client
    return mylib.gst(amount)

df = spark.createDataFrame([(100.0,), (250.0,)], "amount DOUBLE").select(with_gst("amount").alias("gross"))
try:
    df.collect()
except Exception as e:
    print("before addArtifacts:", [l.strip() for l in str(e).splitlines() if "ModuleNotFoundError" in l][:1])
spark.addArtifacts("mylib.py", pyfile=True)
print("after addArtifacts: ", [r.gross for r in df.collect()])
# before addArtifacts: ["ModuleNotFoundError: No module named 'mylib'"]
# after addArtifacts:  [118.0, 295.0]
spark.stop()

addArtifacts uploads the file through the AddArtifacts RPC into a directory private to the session, and puts it on the Python path of that session’s workers. archive=True does the same for a packed environment, and file=True for plain files.

Good to know: the rule of thumb is that the client needs only what builds plans (pyspark-client and your own code), and the server needs everything that runs data: the right Python version, and any package a UDF imports, either installed in the server image or uploaded as an artifact. Artifacts are per-session, so another client’s session never sees your modules.

Controlling long-running work

How it works. Every execution is an operation with an ID, and a session can attach tags to the operations it starts. Another thread, or another client holding the same session, can cancel operations by tag:

import threading, time
from pyspark.sql import SparkSession

spark = SparkSession.builder.remote("sc://sc-server:15002").getOrCreate()
result = {}

def long_job():
    spark.addTag("nightly-etl")
    try:
        spark.range(0, 50_000_000_000).selectExpr("sum(id % 7)").collect()
        result["status"] = "finished"
    except Exception as e:
        result["status"] = str(e).splitlines()[0][:120]

t = threading.Thread(target=long_job); t.start()
time.sleep(6)
print("interrupted operations:", len(spark.interruptTag("nightly-etl")))
t.join()
print("job ended with:", result["status"])
# interrupted operations: 1
# job ended with: (org.apache.spark.SparkSQLException) [OPERATION_CANCELED] Operation has been canceled. SQLSTATE: HY008
spark.stop()

interruptTag returned the one operation it cancelled, and the blocked collect() in the other thread ended with OPERATION_CANCELED. interruptAll() cancels everything in the session.

Operations also survive a broken connection. With reattachable execution, on by default (spark.connect.execute.reattachable.enabled), the server keeps a running operation and a buffer of its unread results (observerRetryBufferSize, 10 MB) when the client’s stream drops. The client reattaches with ReattachExecute and carries on from where it stopped reading. From 4.1, resending the same ExecutePlan with the same operation ID also reattaches instead of running the query twice.

Good to know: tags are per thread in the Python client, which is why the tag is added inside long_job. That lets a web service tag each request’s queries with its request ID and cancel exactly those when the request is abandoned.

Structured Streaming over Connect

How it works. A streaming query started from a client runs on the server. What the client gets back is a handle; its methods (status, lastProgress, isActive, stop()) are RPCs to the server:

import time
from pyspark.sql import SparkSession

spark = SparkSession.builder.remote("sc://sc-server:15002").getOrCreate()
q = (spark.readStream.format("rate").option("rowsPerSecond", 20).load()
       .selectExpr("value % 3 AS bucket").groupBy("bucket").count()
       .writeStream.outputMode("complete").format("memory").queryName("buckets")
       .trigger(processingTime="2 seconds").start())
print("handle class:", type(q).__module__ + "." + type(q).__name__)
while not (q.lastProgress and q.lastProgress["batchId"] >= 2):
    time.sleep(2)
p = q.lastProgress
print("active:", q.isActive, "| batchId:", p["batchId"] >= 2, "| has durationMs:", "triggerExecution" in p["durationMs"])
print("buckets:", sorted(r.bucket for r in spark.sql("SELECT * FROM buckets").collect()))
q.stop()
print("active after stop:", q.isActive)
# handle class: pyspark.sql.connect.streaming.query.StreamingQuery
# active: True | batchId: True | has durationMs: True
# buckets: [0, 1, 2]
# active after stop: False
spark.stop()

Good to know: this cost me a server. My first version of this script crashed on the client, before it reached q.stop(), and the query kept running on the server: its run ID was still doing state-store maintenance in the server log while my next attempt started a second query beside it. The two, on a small container, ran the server out of memory and Docker killed it. A streaming query belongs to the server, not to the script that started it. Stop queries in a finally block, and list what is running with spark.streams.active when you reconnect. foreachBatch functions and streaming listeners written in Python run in a Python process on the server for the same reason.

What the thin client cannot do

How it works. Anything that needs direct access to the JVM does not exist on the client, and says so with a specific error class:

from pyspark.sql import SparkSession

spark = SparkSession.builder.remote("sc://sc-server:15002").getOrCreate()
df = spark.range(3)
for label, fn in [("spark.sparkContext", lambda: spark.sparkContext),
                  ("df.rdd", lambda: df.rdd),
                  ("df._jdf", lambda: df._jdf)]:
    try:
        fn(); print(label, "-> works")
    except Exception as e:
        print(f"{label:20s}-> {type(e).__name__}: {str(e).splitlines()[0][:110]}")
spark.stop()

spark.sparkContext  -> PySparkAttributeError: [JVM_ATTRIBUTE_NOT_SUPPORTED] Attribute `sparkContext` is not supported in Spark Connect as it depends on the JVM. ...
df.rdd              -> PySparkAttributeError: [JVM_ATTRIBUTE_NOT_SUPPORTED] Attribute `rdd` is not supported in Spark Connect as it depends on the JVM. ...
df._jdf             -> PySparkAttributeError: [JVM_ATTRIBUTE_NOT_SUPPORTED] Attribute `_jdf` is not supported in Spark Connect as it depends on the JVM. ...

With sparkContext go RDDs, sc.broadcast, accumulators, sc.setJobGroup, sc.addFile and any library that reaches into the JVM through _jvm or _jdf.

Good to know: most RDD code has a DataFrame equivalent, and the gap has been closing release by release. mapInPandas and mapInArrow replace rdd.mapPartitions; applyInPandas replaces most groupByKey logic; tags and interruptTag replace job groups; addArtifacts replaces addFile; and 4.2 adds zipWithIndex and toJSON on DataFrames. Code that must call into the JVM directly, such as some older connector libraries, needs a classic session.

Running existing code over Connect: spark.api.mode

How it works. From 4.0, an ordinary PySpark program can switch to Spark Connect with one setting and no code changes. With spark.api.mode=connect and a local master, getOrCreate() starts a Connect server inside the same machine and returns a Connect session connected to it:

from pyspark.sql import SparkSession

spark = (SparkSession.builder
         .master("local[2]")
         .config("spark.api.mode", "connect")      # start a local Connect server behind the scenes
         .getOrCreate())
print("session class:", type(spark).__module__ + "." + type(spark).__name__)
print(spark.range(4).selectExpr("sum(id) AS s").first().s)
# UserWarning: Failed to set spark.api.mode to Some(connect) due to [CANNOT_MODIFY_CONFIG] ...
# session class: pyspark.sql.connect.session.SparkSession
# 6
spark.stop()

The program asked for a local master and got a Connect session. The CANNOT_MODIFY_CONFIG warning is harmless: the client tries to copy every builder setting into the new remote session, and spark.api.mode is a static setting the server has already consumed.

Good to know: this is the cheapest way to find out whether an existing codebase is Connect-compatible: run its test suite with spark.api.mode=connect and look for JVM_ATTRIBUTE_NOT_SUPPORTED errors. It needs the full pyspark package, since the server runs locally, plus the Connect client’s dependencies (grpcio, grpcio-status, protobuf, googleapis-common-protos).

Securing a Connect server

How it works. The open-source server has no authentication by default: spark.connect.authenticate.token is undefined in 4.1.3, and anyone who can open a connection to port 15002 can run queries, read any table the server can read, and run arbitrary Python through a UDF. Setting a token makes the server require it as a bearer token on every call:

$SPARK_HOME/sbin/start-connect-server.sh --conf spark.connect.authenticate.token=s3cret

I ran three clients against a 4.1.3 server started with that token, on localhost. The client with no token was refused on its first call:

status = StatusCode.UNAUTHENTICATED
details = "No authentication token provided"

The two clients that did pass a token never got that far. A token makes the client switch to a secure channel, and the plaintext server could not complete the handshake, so the client retried in a loop of Handshake failed with error SSL_ERROR_SSL ... WRONG_VERSION_NUMBER until I stopped it. In other words, the token check on the server works, but a client can only present a token over TLS, and the open-source server does not terminate TLS itself. Token authentication needs a TLS-terminating proxy in front of the server, which is also where most teams want authentication to live.

On the client side, a token in the URL, or the SPARK_CONNECT_AUTHENTICATE_TOKEN environment variable, is what switches the channel to a secure one. The 4.1.3 client’s DefaultChannelBuilder.secure property is literally use_ssl or token is not None, which is why the run above went looking for TLS as soon as a token was present.

Good to know: in practice, teams put the server behind something that does authentication and TLS properly: a gRPC-aware reverse proxy (Envoy, NGINX) or a service mesh sidecar that validates identities, with the server itself listening only on a private network. Whatever the front door, remember that a Connect server executes client-supplied Python, so it deserves the same trust boundary as a shell on the driver host.

Other clients

Scala and Java

The JVM client is a separate artifact, spark-connect-client-jvm, with the Dataset API:

val spark = org.apache.spark.sql.SparkSession.builder().remote("sc://localhost:15002").getOrCreate()
spark.range(5).show()

Scala UDFs and typed operations need their classes on the server, uploaded as artifacts (spark.addArtifact("app.jar")) or installed with the server.

Go, Rust and Swift

Clients in their own repositories under the Apache Spark project (spark-connect-go, spark-connect-rust, spark-connect-swift), covering the DataFrame and SQL APIs. They are what the protocol design makes possible: a Go service can run Spark queries without a JVM anywhere in its image.

JDBC

From 4.1, a JDBC driver speaks the Connect protocol, with URLs of the form jdbc:sc://host:15002, for SQL tools that only understand JDBC.

Operating a Connect server

Watch it in the UI

The server’s Spark UI has a Spark Connect tab (/connect/ on port 4040 by default), with session statistics and every statement each session ran; from 4.2 the history server shows the same tab for finished servers.

Know the limits

The 4.1.3 defaults worth knowing, read from the server’s own configuration entries:

  • spark.connect.grpc.binding.port is 15002.
  • spark.connect.grpc.maxInboundMessageSize is 134217728b (128 MB), the largest single message a client can send, which bounds plans and uploaded chunks.
  • spark.connect.grpc.arrow.maxBatchSize is 134217728b (128 MB) per result batch sent back.
  • spark.connect.session.manager.defaultSessionTimeout is 60m of idleness before a session is released.
  • spark.connect.execute.reattachable.enabled is true, with a 10 MB observerRetryBufferSize of results kept for reattaching clients.

Mind the memory

Everything the classic driver held for one user, the server holds for everyone: query plans, broadcast tables, results being streamed back, streaming queries. Size its memory for concurrent users, not for one, and keep collect()s of large results out of shared servers; toLocalIterator() streams results in batches instead.

Pin the client to the server version

Install the client at exactly the server’s version where you can. The setup section shows the two directions: a 4.1.3 client against a 3.5.9 server ran range and failed on createDataFrame and on every spark.sql call, while a 3.5.9 client against a 4.1.3 server ran the whole script. So when versions must differ during a migration, upgrade the server first and let older clients follow; never run a client newer than its server. One script is not a compatibility guarantee, so run your own test suite across any pair you intend to support.

What it replaced, and why a new protocol

Remote access to Spark is not new. Three designs came before Spark Connect, and each solved part of the problem.

Py4J inside the client process. Classic PySpark. It gives the full API, RDDs included, at the cost of a JVM and a whole driver in every client, which is the coupling this post started with.

The Thrift JDBC/ODBC server. One long-running Spark application that speaks the HiveServer2 protocol. It decouples clients from the driver, but only for SQL strings: no DataFrame API, no Python UDFs from the client, and every connection shares one session model built for BI tools.

Apache Livy. A REST service that holds interactive sessions and runs code snippets or batch jobs submitted as text. It also moves the driver out of the client, but what crosses the wire is code, Scala or Python source to be interpreted on the server, so the client has no DataFrame objects, no type checking and no IDE support, and results come back as serialised text.

Spark Connect’s design decision was to send plans instead of code or SQL strings. A plan is structured, so the client can offer the real DataFrame API with autocompletion and types; it is declarative, so the server can analyse and optimise it exactly as a local driver would; and it is language-neutral, so every client builds the same messages. The cost of that decision is that anything which is not a plan, arbitrary JVM calls above all, cannot cross.

Advantages and trade-offs

The same query, the same optimizer and the same executors on both sides; what differs is where each concern lives:

Concern Classic session Spark Connect
Where the driver runs Inside your process In a shared server process
Client install Java plus full pyspark (455 MB) pyspark-client (1.6 MB), no Java
What crosses the boundary Py4J calls to a local JVM Protobuf plans out, Arrow batches back
When a bad column is reported When the DataFrame is defined When a schema or result is first needed
RDDs, sparkContext, _jdf Available JVM_ATTRIBUTE_NOT_SUPPORTED
Where UDF code runs Python workers of your own driver Python workers of the server, same Python minor version required
A client crash Kills the driver and its jobs Kills the client; batch work is cleaned up, streams keep running
Upgrading Spark Every client at once Server and pinned clients together
Authentication Your process, your credentials None by default; add TLS and a token or a proxy

What it buys

  • Thin clients. A 1.6 MB pip install instead of Java and 455 MB of Spark, which makes Spark usable from IDEs, notebooks, web services and small containers.
  • Stability and isolation. A client that crashes or runs out of memory takes down only itself. The driver’s memory is no longer shared with user code.
  • Independent upgrades. The server can be upgraded without redeploying every client, and clients’ own dependencies no longer collide with Spark’s.
  • Many languages. Any language that speaks gRPC and Arrow can be a client: Python, Scala, Java, Go, Rust, Swift and JDBC today.
  • Multi-tenancy. Many users share one driver and one cluster, each with an isolated session.
  • Debuggability. Clients can be run and stepped through in an ordinary IDE, with no JVM attached, while the server is inspected separately.

What it costs

  • No JVM-level APIs. RDDs, sparkContext, broadcast variables, accumulators and libraries that use Py4J directly do not work.
  • A shared server is a shared failure domain. Moving the driver out of clients concentrates it in one process; that process must be sized, monitored and made highly available.
  • Latency per round trip. Schema lookups and small actions are network calls; code that interleaves many small operations is slower than on a local driver.
  • Environment on two sides. Python versions and UDF dependencies must match on the server, which is a new place for “works on my machine” to fail.
  • Security is your job. No authentication by default, and a server that executes client code.

When should you use it?

Use Spark Connect for interactive and multi-user access to a shared cluster (notebooks, BI and SQL tools, data apps), for services that run Spark queries on behalf of users, for thin or non-JVM environments, and for new pipelines that you want insulated from Spark upgrades. Declarative Pipelines and other new Spark components are built on it, so new code written against it has a longer runway.

Keep a classic session for batch jobs that already work and gain nothing from a server, for code that depends on RDDs or on libraries that call into the JVM, and for workloads that are extremely sensitive to per-call latency, such as tight loops of tiny queries.

Common misconceptions

“Spark Connect runs Spark on my laptop.” The opposite. The client runs on your laptop; Spark, meaning the driver and the executors, runs wherever the server is.

“My code runs on the client, so my Python packages only need to be on the client.” Only the code that builds plans runs on the client. UDFs, mapInPandas functions and foreachBatch functions run on the server, so their Python version and packages matter there.

“If the client exits, the work stops.” Batch operations are cleaned up, but a streaming query started from a client keeps running on the server until it is stopped, which is how I lost a server while writing this post.

“Spark Connect is slower.” Once a query reaches the server it runs exactly as it would in a classic session: same optimizer, same executors. The costs are per-round-trip latency and result transfer, which matter for chatty code and huge collect()s, not for the query itself.

“A Connect server is secure because it is internal.” It has no authentication until you add it, and it runs arbitrary Python sent by clients.

Frequently asked questions

Which Spark version do I need? The Python client arrived in 3.4, streaming and the Scala and Go clients in 3.5, and the lightweight pyspark-client package and spark.api.mode in 4.0. For new work, use a 4.x server and a matching client.

Do I need Java on the client? No. pyspark-client is pure Python; the example container in this post never had Java installed.

How do I switch an existing PySpark job to Connect? Replace .master(...) with .remote("sc://host:15002"), or keep the code and set spark.api.mode=connect. Then fix anything that raises JVM_ATTRIBUTE_NOT_SUPPORTED.

Why does my UDF fail with PYTHON_VERSION_MISMATCH? Because it runs in Python workers on the server, and the client and server have different Python minor versions. Use the server’s Python version on the client.

Why does ModuleNotFoundError appear for a module I can import locally? The module is on the client, and the UDF that imports it runs on the server. Upload it with spark.addArtifacts("mod.py", pyfile=True), or install it in the server image.

How do I cancel a query from another thread? Tag it with spark.addTag("name") in the thread that runs it, and call spark.interruptTag("name") from anywhere holding the session.

Can several people share one server? Yes; each client gets its own session with its own temporary views and settings, sharing the cluster. Use the fair scheduler to keep one user’s work from starving the others.

The mental model

Spark Connect does not change how Spark executes a query. It changes where the driver lives and what crosses the boundary: plans go one way, Arrow batches come back, and everything that needs the JVM stays on the server. Almost every behaviour in this post, the late analysis errors, the per-session views, the UDFs that fail on Python versions, the streaming query that outlived its script, follows from asking which side of that boundary a piece of code runs on.

References

Trademarks

Apache Spark, Apache Arrow, Apache Iceberg, Apache Hudi and Apache are either registered trademarks or trademarks of The Apache Software Foundation in the United States and other countries.

Found this useful?

These posts and tools are free. If one saved you an afternoon, you can buy me a coffee.

Buy me a coffee