從 Spark Connect 看 Spark 4.x 各模組的發展方向

Spark is moving its centre of gravity out of the JVM. Spark Connect, introduced in 3.4, lets a client drive Spark over gRPC without running a JVM of its own, and in the 4.x line that thin client has become the thing new features are built for: pip install pyspark-client gets you the DataFrame API with none of the JARs full PySpark carries. That gives a simple way to tell which parts of Spark have a future. The state of each Spark module is best read through a single test: can you reach it from a thin Python client over Spark Connect? The modules getting sustained investment were either built for that world (SQL, Declarative Pipelines) or are being rebuilt to reach it (Structured Streaming, MLlib, the Python UDF path); the ones fading (GraphX, SparkR, DStreams, the RDD-based spark.mllib) are exactly the ones that can't cross it.

I learned to read Spark this way by accident. In February I tested Real-Time Mode on open-source Spark 4.1, and the first obstacle wasn't latency; it was Python. The real-time trigger existed in Scala, PySpark had no way to ask for it, and to run the pipeline at all I wrote a helper that reached through the Py4J gateway and built the Scala trigger object by hand:

def with_real_time_trigger(dsw, checkpoint_interval="5 minutes"):
    jvm = dsw._spark._jvm
    jTrigger = jvm.org.apache.spark.sql.execution.streaming.RealTimeTrigger.apply(
        checkpoint_interval
    )
    dsw._jwrite = dsw._jwrite.trigger(jTrigger)
    return dsw

It worked, and it's exactly the kind of code that has no future. It only runs in classic mode, because a Spark Connect client has no JVM to reach into; the PySpark 4.2 Connect session refuses _jvm by name, along with _jsc, sparkContext, and the rest of the Py4J surface. Five months later the helper was obsolete anyway. Spark 4.2 added trigger(realTime=...) to PySpark, so the same pipeline runs from a thin client, and the 4.3 branch lets that Python code run aggregations and deduplication in real time too. A feature that existed but couldn't be reached from Python became real the moment it crossed the client boundary, and that is the story of nearly every module below.

Crossing isn't free, and part of this post is what it cost me when I tried. It's written against Spark 4.2.0 (released July 14, 2026). Spark 4.3 is close but not out: its first release candidate was tagged on August 31, and the vote on it failed on September 6, so where 4.3 matters I describe what's on the release branch, not what has shipped.

What "reachable over Connect" actually means

Spark Connect splits the application from the engine. The client builds an unresolved logical plan, serializes it as protobuf, sends it over gRPC, and gets results back as Arrow batches; everything that plans and runs the query lives on the server.

That contract is the whole test. Anything expressible as a logical plan travels; anything that assumes the client and the driver share a JVM doesn't. The line falls in a predictable place:

RDDs are the dividing line because an RDD isn't a plan; it's a handle to partitioned data and closures living in the driver's JVM. Every module built on RDDs inherits that limitation, which is most of the reason the left-hand column of the diagram above looks the way it does.

The payoff for crossing the line is a client with no JVM. pyspark-client ships only the gRPC client, the protobuf definitions, and the DataFrame API; it installs none of the roughly 277 jars that full pyspark carries. When I measured the 4.2.0 packages, Spark's own Python payload dropped from roughly 480 MB to about 20 MB and the jar directory to zero.

The common "1.5 MB client" line is misleading, though: 1.60 MB is the wheel, and a working thin-client virtualenv still came to roughly 360 to 380 MB across my installs, because pyarrow, pandas, and numpy are hard requirements and dwarf Spark itself. The honest headline is no JVM, no JDK, no jars, not small.

The classic/Connect choice is controlled by spark.api.mode, whose default is still classic; the proposal to make Connect the default (SPARK-50411) is open. So the direction is clear, but for now you opt in, and what that means for someone choosing today is simple: anything you write that touches sparkContext or _jvm is a bet against where the project is going.

Built for it: SQL and Declarative Pipelines

Spark SQL passes the test almost by definition, because a Connect request is a logical plan and Spark SQL is the machinery that turns logical plans into execution. Every DataFrame call, every spark.sql() string, and every Structured Streaming query goes through the same pipeline:

That's why SQL keeps absorbing features that used to live in application code. Across the 4.x line it gained procedural scripting (generally available in 4.1, with cursors added in 4.2), recursive common table expressions (a request that sat open as SPARK-24497 from 2018), the VARIANT type for semi-structured data (shredded into typed Parquet columns from 4.1), collations, and pipe syntax. None of these needs anything from the client beyond the text of the query, which is the point: the more logic moves into SQL, the less it matters what language or process the client is.

-- Recursive CTE: walk an org chart without a loop in application code
WITH RECURSIVE reports (id, name, manager_id, depth) AS (
  SELECT id, name, manager_id, 0 FROM employees WHERE manager_id IS NULL
  UNION ALL
  SELECT e.id, e.name, e.manager_id, r.depth + 1
  FROM employees e JOIN reports r ON e.manager_id = r.id
)
SELECT * FROM reports;

The most interesting 4.2 addition, and the one I spent the most time probing, is metric views (SPARK-54119): a semantic layer declared with CREATE VIEW ... WITH METRICS, where measures such as a conversion rate are defined once and re-aggregated correctly at whatever grain you query.

The problem they solve is real. In an example I built on synthetic session data, the true monthly conversion rate was 0.0923, while the naive average of the five regional rates was 0.2384, 2.6 times too high, because an average of ratios gives a region with 200 sessions the same weight as one with 50,000. A metric view makes that mistake hard to write.

What it doesn't yet do is fail gracefully.

I ran a grammar probe of 87 statements against the official apache/spark:4.2.0 image (once in classic mode and twice over Connect, all three runs identical), and six invalid statements didn't produce a clean error; they crashed the planner with INTERNAL_ERROR, the message that begins "You hit a bug in Spark":

The numbers are protected; the person writing the query isn't, yet.

Declarative Pipelines (SDP) extend the same idea from queries to whole pipelines: you declare what each table should contain, and Spark resolves dependencies, orders execution, and manages checkpoints and writes. Pipelines are defined in Python or SQL and run over Spark Connect by design, so SDP was born on the right side of the line.

It arrived in 4.1; 4.2 added AUTO CDC for applying change feeds as SCD Type 1, and the 4.3 branch adds SCD Type 2 end to end and a SQL form, CREATE FLOW ... AS AUTO CDC. I tested the 4.2 behaviour in detail in AUTO CDC: Turning an event log into current state.

For someone choosing today: push as much logic as you can into SQL and declarative definitions. It's the part of Spark that works identically from every client, it's where new features land first, and it's the part least likely to strand you in a migration.

Rebuilt to reach it: streaming, ML, and Python UDFs

The second group are modules that predate Spark Connect and are being reworked to pass the test. Each has made real progress, and each still has a visible seam.

Structured Streaming and Real-Time Mode

Spark has had three streaming engines, and the history is a clean illustration of the thesis. DStreams (described by Matei Zaharia and colleagues in 2012 and 2013) modelled a stream as a sequence of small RDDs; Structured Streaming (Spark 2.0, 2016; written up by Michael Armbrust and colleagues at SIGMOD 2018) rebuilt streaming on DataFrames, as an unbounded table queried with the same API as batch; and Real-Time Mode, added in 4.1, replaces the micro-batch loop with a continuous pipeline:

Real-Time Mode is also where I learned how wide the gap between a feature's design and its open-source reality can be. The widely quoted single-digit-millisecond figures come from Databricks' own deployments.

On open-source 4.1, a stateless Kafka-to-Kafka passthrough I measured (500 events per second on a single 16-core machine, six runs of roughly two thousand events each) had a p50 of about 17 ms and a p99 of 21 to 26 ms. That's fast; it's also a different number, measured on a much smaller setup, and I'd treat both figures as specific to where they came from. The limits mattered more than the latency:

• Stateless only. The 4.1 operator allowlist permitted projections, filters, and broadcast joins; aggregations, shuffles, and state store operators were off it.
• No Python UDFs. BatchEvalPythonExec wasn't on the allowlist.
• No PySpark trigger, hence the Py4J helper at the top of this post.
• Every stage runs concurrently, so a 16-partition Kafka source needs 16 task slots just to read.

Spark 4.2 fixed the Python gap (the realTime trigger in PySpark) but, per the 4.2 source, left the operator allowlist unchanged, so on 4.2 it's still a stateless engine:

# Spark 4.2+: Real-Time Mode from PySpark, no JVM access needed.
# The argument is the checkpoint interval, not a latency target.
query = (
    events.select(from_json(col("value").cast("string"), schema).alias("e"))
    .select("e.*")
    .filter(col("amount") > 0)                      # stateless: allowed
    .selectExpr("to_json(struct(*)) AS value")
    .writeStream.format("kafka")
    .option("kafka.bootstrap.servers", "kafka:9092")
    .option("topic", "payments_clean")
    .option("checkpointLocation", "/checkpoints/payments_clean")
    .outputMode("update")
    .trigger(realTime="5 minutes")
    .start()
)

The 4.3 branch is where it becomes a general engine. It adds stateful Real-Time Mode (umbrella SPARK-54699): streaming aggregations including windowed ones, dropDuplicates, the JVM transformWithState, and pipelined shuffles, and it removes the @Experimental marker from the trigger APIs.

The branch documentation is just as clear about what remains out of scope: stream-stream joins, flatMapGroupsWithState, session windows, and the PySpark transformWithState APIs are unsupported; Python UDFs are still not allowed (SPARK-57237 is open); Kafka is the only source; and stateful queries need enough cores to run both stages at once.

Notice where the seam is. When I built the 4.3 branch and ran it against my own stack, the stateful operators that are ordinary DataFrame operations worked from PySpark in Real-Time Mode: a streaming aggregation and a dropDuplicates both ran at about 40 ms median latency (against about 600 ms for the same query as a one-second micro-batch), with no duplicates getting through. What's still missing is Python's own stateful API: PySpark transformWithState isn't supported in Real-Time Mode, and on the branch today it fails with a misleading checkpoint-metadata error (STDS_FAILED_TO_READ_OPERATOR_METADATA) rather than the documented unsupported-operator one. Custom state in real time is still a Scala feature.

The rest of Structured Streaming is further along. transformWithState (SPARK-46815) replaced the older mapGroupsWithState APIs in 4.0 with typed, evolvable state, and it has a Python API, which ended the era when complex stateful streaming meant writing Scala. RocksDB is the default state store in 4.x, and 4.1 enabled adaptive query execution for stateless streaming queries (SPARK-53941).

For someone choosing today: on 4.2, use Real-Time Mode from Python for stateless transforms where the latency matters, and micro-batch Structured Streaming for anything stateful. On 4.3, aggregations and deduplication move into Real-Time Mode from Python as well; if you need custom state (transformWithState) at real-time latency, that's still Scala.

MLlib on Connect

MLlib's problem was more fundamental than streaming's. In classic PySpark, model.fit(df) returns a model object that lives in the driver's JVM, and the Python client just holds a Py4J reference to it. A Connect client has no JVM to hold anything, so ML over Connect required moving model state to the server:

The server keeps fitted models in a cache and hands the client an ID; fit, transform, evaluation, attribute lookup, and save/load become RPCs. This is the one rebuild that's essentially finished. pyspark.ml on Connect shipped in 4.0 (SPARK-50812); 4.1 brought it to general availability with model cache offloading, model size and cache limits, and summaries that survive eviction (SPARK-51236); 4.2 re-enabled FPGrowth (SPARK-54574); and the 4.3 branch adds model-size estimation for around twenty model types (umbrella SPARK-58181) plus a fix for size overcounting (SPARK-57521).

What it hasn't done is grow: the estimators, transformers, and evaluators registered for Connect are the same list from 4.0 through the 4.3 branch, a few known gaps remain open (SPARK-51473, SPARK-51478), and the ML guide still has no section on Connect at all.

Meanwhile the DataFrame-based spark.ml has been the primary API since Spark 2.0, when the RDD-based spark.mllib entered maintenance mode; spark.mllib doesn't pass the Connect test and isn't going to. For deep learning, the path is TorchDistributor (SPARK-41589), which uses Spark to launch distributed PyTorch training rather than reimplementing it.

For someone choosing today: spark.ml pipelines over Connect are safe for the algorithms MLlib already has; don't expect the catalogue to expand, and use TorchDistributor or an external framework for anything modern.

Python UDFs on Arrow

The Python UDF path is the rebuild I've measured most closely. Its history is a long march away from pickling one row at a time: Pandas UDFs (2018) moved data in Arrow batches but converted them to pandas; useArrow=True (from 3.5) put ordinary scalar UDFs on Arrow transport; and @arrow_udf (public in 4.1, SPARK-53014) hands your function a pyarrow.Array with no pandas conversion at all.

In my benchmark on Spark 4.1, what decided the cost of a UDF was mostly what the function received, not how the bytes travelled. useArrow=True was worth 1.4x to 2.4x over pickle depending on the workload; the vectorized UDFs were faster again; and @arrow_udf beat @pandas_udf by about 1.6x on strings but only 5 to 9% on numeric work, because a numeric pandas Series is cheap to build from Arrow while strings are not. The details, and a correction to my own benchmark, are in Stop Blaming Serialization for Slow Python UDFs.

Getting Arrow on by default took two tries. SPARK-48516 flipped the default for Spark 4.0, and Hyukjin Kwon reverted it in March 2025, before 4.0 shipped. Spark 4.2 finally turned it on (SPARK-54555), for regular Python UDFs, UDTFs, and the pandas conversions, and raised the minimum PyArrow to 18.0.0.

The 4.3 branch drops Python 3.10 (SPARK-56889) and adds an opt-in pipelined UDF transfer (spark.python.udf.pipelined.enabled, off by default, SPARK-56642); pandas 3 still isn't fully supported even there (SPARK-55139 is open).

The same Arrow-first direction shows up in the Python Data Source API (SPARK-44076), which lets you write a connector entirely in Python and, since 4.1, receive pushed-down filters. That removes one of the last reasons a Python team needed Scala.

For someone choosing today: on 4.2, plain @udf already gets Arrow transport for free; write new UDFs as @arrow_udf when the logic maps onto pyarrow.compute (especially strings), and @pandas_udf when you need pandas or SciPy.

What crossing the boundary cost me

Everything above argues that Connect is where Spark is going. It's worth being equally concrete about the price of getting there, because I paid it in two ways this month.

Migrating a real pipeline. I moved the pipelines in my lakehouse-stack project to Spark Connect and started with an audit script over the repository. It reported 27 findings across 24 files, and 21 of them were the same line: spark.sparkContext.setLogLevel("WARN").

That single call was the entire blocking surface, and it has no client-side equivalent, because the log level belongs to the server's JVM; the fix is to configure logging where the Connect server starts and delete the call.

The migration itself was almost mechanical, and after it all ten tables matched their pre-migration versions on schema, row count, and an order-independent content hash (that parity result is recorded in my migration notes rather than as a saved log).

The real blocker was the ecosystem. Apache Iceberg doesn't publish a Spark 4.2 runtime; the newest on Maven Central targets 4.1. And the 4.1 jar can't simply be reused, because Spark 4.2 changed the catalog interfaces it implements: View became a class, so loading the old runtime fails with

java.lang.IncompatibleClassChangeError: class org.apache.iceberg.spark.source.SparkView
can not implement org.apache.spark.sql.connector.catalog.View, because it is not an interface

and because that failure happens while the catalog loads, it takes down tables, reads, and writes along with views. 4.2 also requires catalogs that serve both tables and views to implement a new RelationCatalog interface. I built a local Iceberg runtime for 4.2 to keep going, which is fine for a demo and not something I'd run in production.

The general lesson is that a Spark minor version can break connectors, and your upgrade schedule is set by your slowest connector, not by Spark.

Mixing versions. Connect's promise is that client and server can be upgraded independently. I tested all four combinations of 4.1.2 and 4.2.0 clients and servers, running each fourteen-operation matrix twice, with identical results:

The createDataFrame failure is worth spelling out, because it's so small. A 4.2 client checks the size of local data against a limit it reads from the server (SPARK-55047); a 4.1.2 server reports that limit as 3221225472b, with a unit suffix, and the new client's integer conversion fails on the b:

ValueError: invalid literal for int() with base 10: '3221225472b'

So a new client can't build a three-row DataFrame on an old server. Independent upgrades hold in one direction only: upgrade the server first, and pin clients to it.

Left behind: GraphX, SparkR, and the RDD APIs

The last group fails the test for the same structural reason: each is built on RDDs or on a pre-Connect bridge, and none has a path across.

GraphX was one of Spark's original libraries (Spark 0.9, 2014), built entirely on RDDs and never given a Python API. Between September 30 and October 8, 2024, the dev list voted on Holden Karau's proposal to officially deprecate it in Spark 4; she noted that it "hasn't been updated meaningfully in several years" and was already "de-facto deprecated," and was explicit that it was "not a vote to remove GraphX." The proposal passed, although the 4.2.0 GraphX programming guide still carries no deprecation notice.

The clearest verdict came from outside the project. GraphFrames, the DataFrame-based graph library that Berkeley and Databricks released as an external package in 2016, had gone quiet for years; its maintainers revived it, moved it to the io.graphframes namespace, added Spark 4 support, and, starting with 0.10.0, stopped depending on Spark's GraphX at all, shipping their own "internal modified and updated fork of the Apache Spark GraphX" instead.

Because GraphFrames stores vertices and edges as DataFrames, graph queries go through Catalyst, work from Python, and pass the Connect test's premise:

# pip install graphframes-py==0.12.2
# pyspark --packages io.graphframes:graphframes-spark4_2.13:0.12.2
from graphframes import GraphFrame

g = GraphFrame(vertices, edges)          # two DataFrames: id, and src/dst
ranks = g.pageRank(resetProbability=0.15, maxIter=20)

# Motif finding: patterns become DataFrame joins that Catalyst can plan
triangles = g.find("(a)-[]->(b); (b)-[]->(c); (c)-[]->(a)")

SparkR was contributed in Spark 1.4 (2015) by Shivaram Venkataraman and the AMPLab team. Hyukjin Kwon committed its deprecation (SPARK-49347) in August 2024, and since 4.0 starting a SparkR session prints a warning that it "is deprecated from Apache Spark 4.0.0 and will be removed in a future version." It uses its own JVM bridge and was never ported to Connect, and most R users had already moved to sparklyr, which is maintained by Posit and supports Spark Connect.

DStreams were deprecated in 3.4 (SPARK-42075), and spark.mllib has been in maintenance mode since 2.0; both are RDD APIs with DataFrame successors that do pass the test.

For someone choosing today: don't start anything new on these. Use GraphFrames for graphs (it's the only option from Python), sparklyr for R, Structured Streaming for streams, and spark.ml for ML, and treat existing code on the old APIs as migration work with a deadline you don't control.

Choosing, module by module

The part of this that surprised me most is how little of the sorting is about quality. GraphX isn't fading because its algorithms were bad, and SparkR isn't fading because R users were wrong; they're fading because they sit on the wrong side of a client boundary the project has decided to build everything else around.

My Py4J helper was on that side too; it still works in classic mode, and on 4.2 nothing needs it.

Further reading

• SPARK-54633: Quarterly release cadence SPIP.