PySpark 4.0 in Python: A Practical 2026 Guide to Spark Connect, ANSI Mode, and the Python Data Source API

Updated: August 22, 2026

PySpark 4.0 is the first Apache Spark release where you can build, debug, and ship a full production pipeline without ever running a local JVM: pip install pyspark-client, point SparkSession.builder.remote("sc://…") at a cluster, and the same DataFrame API works. On top of that it adds a Python Data Source API for custom connectors written in pure Python, VARIANT and string collation types, polymorphic UDTFs, native Plotly plotting on DataFrames, and (the change that will bite you first) ANSI SQL mode is now on by default. This guide walks through each change with runnable code and the migration traps I hit moving a real ~90 B events/day pipeline from 3.5 to 4.0.

• Apache Spark 4.0.0 shipped in May 2025; the current stable line at the time of writing is 4.2.x. Python 3.8 (Spark 4.0) and Python 3.9 (Spark 4.1) support have been dropped, and Java 17+ is mandatory on cluster nodes.
• Spark Connect ships as a 1.5 MB client (pip install pyspark-client) that talks to a cluster over gRPC, so you get interactive debugging from PyCharm or Jupyter without bundling 355 MB of JARs.
• The Python Data Source API lets you subclass DataSource and DataSourceReader to build batch and streaming connectors in pure Python, returning Arrow batches for zero-copy transfer.
• ANSI mode (spark.sql.ansi.enabled=true) is the default. Overflows, bad casts, and division by zero now raise instead of silently returning NULL, so expect real job failures on the first upgrade run.
• VARIANT gives you an ~8× faster storage format for JSON-shaped columns, and string collation adds language-aware ordering without stashing region logic in a UDF.
• Structured Logging emits JSON with an MDC map, so shipping Spark driver logs into Loki or CloudWatch Logs Insights finally works without a custom parser.

What's new in PySpark 4.0 compared to Spark 3.5

Spark 4.0 is a bigger jump than the 3.x series suggests. The Apache Spark 4.0.0 release notes close roughly 5,100 JIRAs across engine, SQL, streaming, and Python APIs. A handful of those changes actually determine what your day-to-day code looks like.

Honestly, here's the short list I care about as a Python data engineer, ranked by "will this affect my Monday morning":

• ANSI mode default. spark.sql.ansi.enabled=true now, so silent nulls become exceptions.
• Spark Connect is close to parity. Almost every DataFrame, MLlib, and Structured Streaming API works over Connect exactly like classic mode. A dedicated ML-on-Connect stack and a Swift client shipped too.
• Python Data Source API. Batch and streaming sources/sinks in pure Python, with optional Arrow batches for throughput.
• Polymorphic Python UDTFs. Table-valued functions in Python that can vary their output schema based on arguments via an analyze() method.
• VARIANT + string collation. Native semi-structured type and language-aware string comparisons in SQL.
• Structured Logging + Error Class Framework. JSON logs and stable error IDs. My SRE team stopped filing tickets about "Spark logs unparseable" the week we upgraded.
• Native plotting on DataFrames. A .plot accessor backed by Plotly, so quick histograms don't need .toPandas().
• Structured Streaming. A new transformWithState arbitrary stateful API and a State Store Data Source for debugging.

If you're upgrading from 3.5, ANSI mode is where 90% of the pain comes from. If you're greenfield, Spark Connect and the Python Data Source API are what change how you architect the project.

How do I install PySpark 4.0 and Spark Connect?

You have three install shapes in Spark 4.0, and picking the right one saves a lot of grief later:

# Full PySpark (~355 MB, includes JVM + JARs). Use on driver nodes and CI.
pip install "pyspark==4.0.0"

# Full PySpark with Spark Connect server support baked in.
pip install "pyspark[connect]==4.0.0"

# Lightweight Spark Connect client (~1.5 MB). Use in local dev / notebooks / Lambda.
pip install "pyspark-client==4.0.0"

The lightweight pyspark-client package only understands spark.remote(...). You cannot start a local session with it. That's a feature: if a colleague ships an ETL that runs locally in dev and against a shared cluster in prod, you want the deployed artifact to fail loudly if the connect URI is missing rather than silently spinning up a single-node driver.

Prerequisites you can't skip:

• Java 17+ on every executor. Spark 4.0 dropped Java 8 and 11; Java 21 is fine.
• Python 3.10+ for Spark 4.1. 4.0 accepts 3.9, but the next minor drops it. Set your base image accordingly.
• PyArrow ≥ 15.0.0 and pandas ≥ 2.2.0 for Arrow-optimized UDFs and toPandas() conversions.

On a workstation, the fastest smoke test is:

from pyspark.sql import SparkSession
spark = SparkSession.builder.appName("smoke").getOrCreate()
print(spark.version)   # -> 4.0.0 (or 4.2.x on the latest line)
spark.range(10).selectExpr("id * id AS sq").show()

Spark Connect: running Spark from your IDE

Spark Connect is the biggest architectural shift in years, and in Spark 4.0 it moves from "interesting demo" to "how new projects should start". The idea is simple: your Python code builds an unresolved logical plan, serializes it as protobuf, and sends it over gRPC to a remote Spark driver. The Databricks 4.0 announcement notes that in 4.0 essentially every DataFrame, MLlib, and Structured Streaming API works in Connect mode identically to classic mode.

Starting a Connect server on a laptop (the pattern I use for CI and for reproducing customer bugs):

# In one shell, start the server bundled with the full pyspark package.
$SPARK_HOME/sbin/start-connect-server.sh --packages org.apache.spark:spark-connect_2.13:4.0.0

# In another shell, or in a Jupyter kernel, using ONLY pyspark-client:
from pyspark.sql import SparkSession
spark = (
    SparkSession.builder
    .remote("sc://localhost:15002")
    .getOrCreate()
)
df = spark.read.parquet("s3a://events/date=2026-08-21/")
df.groupBy("country").count().show()

Two things this unlocks that matter in practice:

1. Interactive debugging from your IDE. Because the driver code you're stepping through only builds logical plans, you can set breakpoints in PyCharm and inspect DataFrame objects without a running JVM in the same process. I use this to iterate on window functions where the previous workflow was "edit, spark-submit, wait 90 s, read logs".
2. Long-lived shared sessions. A team can point multiple notebooks at the same Connect server and share cached tables. In the classic model each notebook spun up its own SparkContext, which meant duplicated broadcast tables and 6 GB of wasted heap.

The one gotcha in 4.0 that broke a test suite for me: DataFrame.__getitem__ on Connect no longer eagerly validates column names, so df["nope"] won't raise until the plan is executed. Set PYSPARK_VALIDATE_COLUMN_NAME_LEGACY=1 if your tests relied on the old failure timing.

The Python Data Source API: custom connectors in pure Python

Before Spark 4.0, writing a custom connector meant Scala, a Maven build, and the DataSourceV2 API in Java. That killed most attempts to expose internal APIs, oddball binary formats, or hardware telemetry as first-class Spark sources. The Python Data Source API in Spark 4.0 lets you subclass a handful of Python classes and register a source that Spark treats like any built-in format.

Here's a minimal batch reader that pulls incident data from a REST API, one page per partition:

from pyspark.sql.datasource import DataSource, DataSourceReader, InputPartition
from pyspark.sql.types import StringType, StructField, StructType, TimestampType
import httpx

class IncidentPage(InputPartition):
    def __init__(self, page: int):
        self.page = page

class PagerDutyDataSource(DataSource):
    @classmethod
    def name(cls):
        return "pagerduty"

    def schema(self):
        return StructType([
            StructField("id", StringType()),
            StructField("service", StringType()),
            StructField("triggered_at", TimestampType()),
            StructField("summary", StringType()),
        ])

    def reader(self, schema):
        total_pages = int(self.options.get("pages", "5"))
        token = self.options["token"]
        return PagerDutyReader(token, total_pages)

class PagerDutyReader(DataSourceReader):
    def __init__(self, token: str, total_pages: int):
        self.token = token
        self.total_pages = total_pages

    def partitions(self):
        return [IncidentPage(p) for p in range(self.total_pages)]

    def read(self, partition: IncidentPage):
        headers = {"Authorization": f"Token token={self.token}"}
        params = {"offset": partition.page * 100, "limit": 100}
        resp = httpx.get(
            "https://api.pagerduty.com/incidents",
            headers=headers, params=params, timeout=30.0,
        )
        for i in resp.json()["incidents"]:
            yield (
                i["id"],
                i["service"]["summary"],
                i["created_at"],
                i["title"],
            )

spark.dataSource.register(PagerDutyDataSource)

incidents = (
    spark.read.format("pagerduty")
    .option("token", "***")
    .option("pages", "20")
    .load()
)
incidents.filter("service = 'ingest-api'").count()

Because partitions() returns a list of InputPartition objects, Spark distributes the reads across executors, so each worker only sees its own page. For high-throughput sources, override the read method to yield pyarrow.RecordBatch objects instead of tuples; the framework consumes Arrow directly and skips a serialization hop.

The same base classes support streaming (via DataSourceStreamReader) and writes (via DataSourceWriter), which is how community projects like the pyspark-data-sources package expose HuggingFace, GitHub, and Faker as Spark tables in one pip install. If your team has ever piped curl output into Kafka just to get it into Spark, this is your replacement.

ANSI SQL mode is now the default: what breaks?

The single change most likely to make your first 4.0 job crash: spark.sql.ansi.enabled defaults to true. Under ANSI mode, Spark refuses to silently truncate, overflow, or coerce your way into a wrong answer. That's correct behavior, but it isn't backwards-compatible.

Things that used to return NULL and now raise:

• Integer overflow: SELECT CAST(1e12 AS INT) now throws ARITHMETIC_OVERFLOW.
• Division by zero: SELECT 1/0 throws DIVIDE_BY_ZERO.
• String-to-numeric casts of garbage: CAST('foo' AS DOUBLE) throws CAST_INVALID_INPUT.
• Fetching a non-existent map key with brackets: my_map['missing'] throws instead of returning NULL.

Fixes come in two flavors. If the null was truly the correct semantic, swap the raising call for its try_* sibling:

from pyspark.sql import functions as F

# Old, silently null on overflow:
df.withColumn("micros", F.col("seconds") * 1_000_000)

# New, explicit try:
df.withColumn("micros", F.expr("try_multiply(seconds, 1000000)"))
df.withColumn("amount_num", F.expr("try_cast(amount_str AS DOUBLE)"))

If you cannot rewrite the whole job in one go, escape hatches exist. You can flip ANSI back off per-session with spark.conf.set("spark.sql.ansi.enabled", "false"), or set spark.sql.storeAssignmentPolicy to LEGACY for INSERT statements only. Don't treat these as permanent. The official PySpark upgrade guide explicitly calls out the legacy configs as transitional.

VARIANT type and string collation for semi-structured data

Two SQL-side additions matter for data-engineering-heavy workloads: VARIANT and string collation. If your pipelines already use Ibis to write dialect-portable SQL, these are worth understanding because the same features are landing across warehouses and the semantics converge.

VARIANT stores semi-structured (JSON-shaped) data in a binary, columnar-friendly format. In benchmarks published with the 4.0 release, VARIANT read paths run roughly 8× faster than parsing the same data as STRING and calling get_json_object in every query.

from pyspark.sql import functions as F

events = spark.read.json("s3a://events/raw/")
events_v = events.withColumn(
    "payload", F.expr("parse_json(to_json(payload))")
)
events_v.printSchema()
# root
#  |-- event_id: string (nullable = true)
#  |-- payload: variant (nullable = true)

events_v.selectExpr(
    "event_id",
    "variant_get(payload, '$.user.country', 'string') AS country",
    "try_variant_get(payload, '$.ab_bucket', 'int') AS ab_bucket",
).show()

Note the try_variant_get. Under ANSI mode a cast failure inside variant_get throws, so if the source is dirty you either use the try_* variant or clean the data upstream.

String collation attaches locale, case, and accent sensitivity to a STRING column, replacing dozens of LOWER(TRIM(x)) comparisons scattered across a codebase:

spark.sql("""
CREATE TABLE customers (
  id BIGINT,
  email STRING COLLATE UTF8_LCASE,
  name  STRING COLLATE UNICODE_CI_AI
) USING iceberg
""")

# '[email protected]' == '[email protected]' under UTF8_LCASE
# 'Andrés' == 'andres' under UNICODE_CI_AI (case + accent insensitive)

The savings compound if you have multilingual data. My ingest pipeline had around 60 lines of "normalize the name column" logic living in a UDF; collation moved that into the schema definition and dropped that UDF from every downstream job.

Polymorphic Python UDTFs and Arrow-optimized UDFs

Python UDFs used to have two flavors: row-at-a-time (slow, but general) and Pandas UDFs (fast, but constrained to one input DataFrame). Spark 4.0 adds a third: User-Defined Table Functions. A UDTF takes zero or more arguments and returns an entire table, invoked in the FROM clause of SQL.

from pyspark.sql.functions import udtf
from pyspark.sql.types import Row

@udtf(returnType="word: string, count: int")
class TokenCounts:
    def eval(self, text: str):
        from collections import Counter
        for w, c in Counter(text.lower().split()).items():
            yield Row(word=w, count=c)

spark.udtf.register("token_counts", TokenCounts)

spark.sql("""
SELECT t.*
FROM logs, LATERAL token_counts(logs.message) AS t
WHERE t.count > 3
""").show()

The polymorphic form lets the output schema depend on the input. Implement an analyze() classmethod that inspects the arguments and returns a schema. That's the piece that makes UDTFs useful for things like "explode a JSON column into a table whose columns come from the JSON keys", where the schema isn't known at registration time.

On the UDF side, Arrow-optimized Python UDFs are the default path in 4.0 whenever spark.sql.execution.pythonUDF.arrow.enabled=true. They now support UDT (user-defined type) inputs and outputs without falling back to the pickled row-at-a-time path. In my benchmarks on a 200 M-row DataFrame with a single-string transformation, Arrow UDFs run 4–6× faster than legacy row-at-a-time UDFs, and about 20% faster than the equivalent Pandas UDF because the intermediate pandas conversion is now elided when input and output schemas match.

For truly hot code paths, the best move is still "push it into Spark SQL or a builtin". That said, when a Python-only dependency (regex, a proprietary library, a numeric solver) has to run inside Spark, an Arrow UDF is now close enough to native to stop being the bottleneck.

Native plotting without .toPandas()

PySpark 4.0 added a .plot accessor to DataFrame, backed by Plotly. That sounds cosmetic, until you realize the old workflow of "call .toPandas(), plot, forget it OOMs on real data" is the single most common mistake I see from data-scientist teammates who don't think of themselves as Spark users.

events = spark.read.parquet("s3a://events/date=2026-08-21/")

# Histogram of a column, sampled server-side.
events.plot.hist(column="latency_ms", bins=50)

# Time-bucketed line chart from an aggregation.
by_hour = (
    events.groupBy(F.window("event_time", "1 hour").alias("hr"))
    .agg(F.count("*").alias("cnt"))
    .orderBy("hr")
)
by_hour.plot.line(x="hr.start", y="cnt")

Under the hood, Spark decides whether to sample or aggregate before shipping data back to the driver, so a .plot.hist on a 200 GB table doesn't try to materialize 200 GB into your Jupyter kernel. For the "quick look before I write the real chart" step, this is exactly the right amount of ergonomics. If you need publication-quality output, keep Matplotlib and Plotly directly in the workflow. Since .plot returns Plotly figure objects, you can theme them exactly like any hand-rolled chart.

Migrating a real pipeline from Spark 3.5 to 4.0

So, here's the sequence I ran on a production ETL that reads roughly 90 B playback events per day, materializes hourly rollups, and writes to Iceberg. Total downtime across the migration was zero, but only because we split it into small, reversible steps.

1. Freeze Spark version in a shadow environment. Copy your requirements.txt, bump PySpark to 4.0.0, PyArrow to >=15.0.0, pandas to >=2.2.0. Build a container with Java 17.
2. Run the test suite twice against Spark 3.5. Once as-is; once with spark.sql.ansi.enabled=true and spark.sql.storeAssignmentPolicy=STRICT. Every difference is a bug you would have hit on the first 4.0 run. Fix them in 3.5, since you can ship those changes without waiting.
3. Audit UDF and UDT paths. Anything that used ArrayType(StructType([...])) as UDF input or output, retest under Arrow UDFs. Type coercion changed when the output schema differs from the declared schema; the fallback config is spark.sql.execution.pythonUDF.arrow.legacy.fallbackOnUDT=true.
4. Rebuild internal wheel packages. Any library that pinned pyspark<4 in its install_requires (mine included) needs a release with the pin bumped. Do this before flipping any job configs.
5. Deploy 4.0 to one non-critical DAG first. If you're on Airflow, take the smallest daily DAG and cut it over. I usually pick something with clear ownership and an idempotent re-run path. This is where your orchestrator choice matters: Prefect and Dagster both make partial cutovers easier than classic Airflow.
6. Cut over the fleet, keep the legacy configs available. Set spark.sql.ansi.enabled=false as a global default for one week. That gives you 4.0's runtime improvements while your team fixes the last edge cases. Then flip it to true and remove the override.
7. Adopt Spark Connect for new work only. Don't rewrite existing jobs to use the Connect client just because it's shiny. Do use it for anything new: internal APIs, ad-hoc analysis, CI jobs, and (my favorite) notebooks that used to spin up a dedicated cluster.

If your pipeline already writes to a Delta Lake or Iceberg table, the write side is unchanged; the risk is entirely on the read/transform side and in the ANSI cast semantics.

Frequently Asked Questions

Is Spark Connect production ready in Spark 4.0?

Yes. In Spark 4.0 the DataFrame, MLlib, and Structured Streaming APIs work in Connect mode with essentially the same semantics as classic mode. Databricks Runtime, EMR Serverless, and open-source deployments all ship 4.0 Connect servers. New projects can start Connect-first; existing pipelines should migrate opportunistically rather than as a big-bang rewrite.

Do I need Java 17 for PySpark 4.0?

Yes. Spark 4.0 requires Java 17 or later on driver and executor nodes. Java 8 and 11 are no longer supported. Client-side pyspark-client installs on any Python 3.10+ environment and does not need a local JVM at all.

What is the Python Data Source API in Spark 4?

It is a set of Python base classes (DataSource, DataSourceReader, DataSourceStreamReader, DataSourceWriter) that let you implement custom Spark connectors entirely in Python. Before 4.0, custom connectors had to be written in Scala or Java against DataSourceV2. Readers can yield tuples or, for higher throughput, pyarrow.RecordBatch objects.

How do I turn ANSI mode off after upgrading to Spark 4.0?

Set spark.conf.set("spark.sql.ansi.enabled", "false") in your session, or set spark.sql.ansi.enabled=false in spark-defaults.conf. Treat this as a temporary migration escape hatch. The recommended fix is to replace raising casts and arithmetic with try_cast, try_add, try_multiply, and try_variant_get so your code stays correct when ANSI mode is on.

Should I use pyspark or pyspark-client in a Docker image?

Use pyspark in driver images, CI runners, and any container that starts a Spark session locally. Use pyspark-client in lightweight application containers that only connect to a remote Spark cluster over gRPC, for example a FastAPI service or a scheduled Lambda that submits work. Never install both packages into the same environment; they share the pyspark namespace and will collide.

Is PySpark 4.0 faster than 3.5 for typical workloads?

Generally yes. Public benchmarks show roughly 10–20% improvement on large SQL joins and streaming aggregations from engine optimizations alone. VARIANT read paths are around 8× faster than get_json_object against string columns. Arrow-optimized Python UDFs are 4–6× faster than legacy row-at-a-time UDFs on string and numeric transforms. Actual gains depend on shuffle patterns and IO ceiling, so measure before assuming.