Spark 4, one minor release at a time: what each feature is, its syntax, and a real run

• The core problem and architecture: how the four releases connect
  • The problem: a release note is not a roadmap
  • Four releases, and a faster cadence
  • Three decisions that run through the line
    • 1. SQL becomes the primary surface again
    • 2. One engine from a nightly batch to a hundred milliseconds
    • 3. Python and Connect move the work off the JVM boundary
  • How to read the maturity of a feature
  • The stack, and where each decision lives
• Under the hood: the features, release by release
  • Spark 4.0: the break, and a far more expressive SQL
    • ANSI mode on by default (SPARK-44444)
    • VARIANT, a type for JSON you do not want to schema
    • SQL pipe syntax (SPARK-49555)
    • SQL user-defined functions (SPARK-46057)
    • Session variables (SPARK-42849)
    • String collation (SPARK-46830)
    • The built-in XML data source (SPARK-44265)
    • The Python Data Source API and Python UDTFs (SPARK-44076)
    • Also in 4.0
  • Spark 4.1: pipelines, procedural SQL, and real-time mode
    • SQL scripting (SPARK-54499)
    • Recursive common table expressions (SPARK-24497)
    • VARIANT generally available, with colon paths (SPARK-54454, SPARK-52494)
    • Sketch functions (SPARK-52515, SPARK-54199, SPARK-52407)
    • Parameter markers wherever you need them (SPARK-53573)
    • Arrow-native UDFs (SPARK-52214, SPARK-52979)
    • Spark Declarative Pipelines (SPARK-51727)
    • Structured Streaming real-time mode (SPARK-53736)
  • Spark 4.2: new types, new joins, and a clearer SQL surface
    • Geospatial types: GEOMETRY and GEOGRAPHY (SPARK-51658)
    • The QUALIFY clause (SPARK-31561)
    • NEAREST BY top-K ranking join (SPARK-56395)
    • SET PATH: where unqualified names are looked up (SPARK-54806)
    • Metric views (SPARK-54119)
    • time_bucket and IGNORE NULLS (SPARK-56594, SPARK-55256)
    • Also in 4.2
  • Spark 4.3: stateful real-time mode, semi-structured SQL, and history tracking
    • JSON_TABLE (SPARK-58366)
    • ASOF JOIN (SPARK-58092)
    • UNNEST in the FROM clause (SPARK-58419)
    • VARIANT editing functions (SPARK-57804 and others)
    • TIME(p) and nanosecond timestamps (SPARK-57551, SPARK-56876)
    • Stateful real-time mode (SPARK-58635, SPARK-57228, SPARK-58508)
    • Auto CDC with SCD Type 2 (SPARK-58321, SPARK-58247, SPARK-58261)
    • New functions (SPARK-56820, SPARK-57253, SPARK-58399, SPARK-58295)
    • Faster queries on small data (SPARK-57851, SPARK-57268)
    • Also in 4.3
• Hands-on demo: run every example yourself
  • What you need
  • Download and verify
  • Four things that went wrong first
  • A runner that executes a SQL file
  • One SQL file per release
  • The PySpark examples
  • Real-time mode against Kafka
  • Cleaning up
• Real-world scenarios and production failures
  • 1. The upgrade that turns a green job red
  • 2. A feature that parses and then does nothing
  • 3. A feature that is off until you switch it on
  • 4. A Python environment that is not ready
  • 5. Real-time mode that will not start, or starts and finds no slots
  • Environment traps that look like Spark bugs
• Optimisation, tuning knobs and best practices
  • The settings that gate or shape these features
  • What each feature costs
  • Which upgrades are worth it on their own
  • Best practices
• Summary and key takeaways
• References
• Trademarks

TL;DR

• The line has shipped four minor releases in about seventeen months, and each one has a different centre of gravity: a break with the past plus a far more expressive SQL, then pipelines and procedural SQL, then new data types and new joins, then stateful low-latency streaming and history tracking.
• Eight or so features per release are covered here, chosen because a data engineer meets them in daily work. Each gets four things: what it is, why it exists, the syntax, and the output of a real run.
• Every example ran on the release that introduced the feature: on an official Docker image, and for the SQL and Scala examples on the plain release tarball too. Where a feature is off by default or labelled preview, the section says so and shows the flag.
• Several of the most talked-about features are not on by default or need a connector. The text names the exact setting rather than implying they work out of the box.
• Removals, default changes and upgrade risks are not repeated here. They are covered in the release history post, which this post links to instead of restating.

A Spark release note is written for the project, not for you. The list for one minor release holds hundreds of items, most of them correct and most of them irrelevant to the job you run on Monday. Four releases in, the useful question is no longer “what changed” but “which of these would I actually use, what does it look like, and what does it cost me?”.

This post answers that for the whole fourth major line, release by release. The four releases differ enough that it helps to see them side by side first:

The first column is where every section below is anchored. A feature belongs to the release that introduced it, so you can read “what do I get by moving from 4.1 to 4.3” as the last two sections and nothing else.

Who this is for. Data engineers and platform leads who already run Spark and need to decide what an upgrade is worth. You know DataFrames and SQL. Terms that are specific to a feature, such as a shredded variant or a change feed, are defined where the feature appears.

What this post does not do. It does not list every ticket, and it does not cover removals or changed defaults. Those are the things that stop an upgrade, and they are in the release history post, which also explains the problem each older release was built to solve. Where a feature below is covered in more depth there, the section links to it and keeps its own explanation short.

How the examples were run. Each feature was run on the release that introduced it, and the output you see is the output of that run, not of a documentation page. The setup, the exact commands, a small runner and one SQL file per release are in the hands-on section near the end. Where a feature needed a flag, a connector or a second tool to run at all, that is stated in the feature’s own section, because that is where you will be reading when it matters.

The rest of the post follows the line in order. First the shape of the four releases and the decisions that connect them, then the features themselves, one release at a time, then how to reproduce every example, the ways these features fail in practice, and finally which of them are worth an upgrade on their own.

The core problem and architecture: how the four releases connect

The problem: a release note is not a roadmap

Read the four release notes in order and they look like four unrelated lists. Read them as one line and three decisions run through all of them. Seeing those decisions first makes each feature in the next section a continuation of something instead of a surprise, and it explains why some features arrive in pieces over two or three releases.

Four releases, and a faster cadence

The first two gaps are almost exactly seven months. The third is half that. The preview counts come from the Maven Central listing for spark-core_2.13 and the Apache archive: 4.0.0 had two previews, and no 4.3.0-preview exists in either. That matters for how you read the last section of this post. Features in the first three releases went through preview builds before they were final. The newest release did not, so it has had the least outside testing, and the patch-release stream that follows it is where the first corrections will land. Patch releases already exist for the older three: 4.0.1 to 4.0.5, 4.1.1 to 4.1.4 and 4.2.1.

Three decisions that run through the line

1. SQL becomes the primary surface again

For a decade the work went into the DataFrame API. The fourth line turns back to SQL, and adds things SQL engines have and Spark did not.

Problem. Teams wrote logic in Python or Scala because SQL could not express it, then paid for that in code that analysts could not read or optimise. Design decision. Put the missing constructs into the SQL language itself, so the optimiser sees them. Trade-off. Each new construct is another thing the parser, the analyzer and the optimiser must keep correct, and several arrive behind a flag first. Alternative. Keep SQL small and push the logic into UDFs, which is the route Spark took before and the one that is opaque to Catalyst.

2. One engine from a nightly batch to a hundred milliseconds

Spark’s streaming and batch stories converge on one declarative layer, and the streaming side gets a second, lower-latency mode.

Problem. A pipeline needs an orchestrator for dependency order, a streaming job for freshness, and hand-written merge logic for change data, and the three are written in different dialects by different teams. Design decision. Declare the datasets and let Spark derive the graph, checkpoints and ordering; add a mode for the streaming queries that need sub-second latency. Trade-off. You hand the execution graph to the framework, which is the point and also the cost: when a run behaves oddly, the schedule is not code you wrote. Alternative. Keep an external orchestrator and micro-batch triggers, which is how most Spark pipelines run today and remains the better fit for pipelines that mix many engines.

3. Python and Connect move the work off the JVM boundary

The client and the language runtime change shape, not just the API.

Problem. Python was the most used language and the slowest path through Spark, because every row crossed the JVM boundary through pickling, and the Python package was a full PySpark install even for a client that only talks to a server. Design decision. Move Python to Arrow end to end, make the client thin (4.0 added a 1.5 MB pyspark-client), and make a server reachable by JDBC and REST. Trade-off. A default that changes how data crosses the boundary changes behaviour at the edges: types, nulls and timestamps are where it shows. Alternative. Keep the pickle path, which is why 4.2 is the release to test on your own UDFs before you trust the default.

How to read the maturity of a feature

Every section below tags a feature with one of these, and they are not interchangeable. The tag comes from the release note, and from whether the example needed a flag to run.

The last row is the one that surprises people. A feature that exists in Spark SQL does not exist for your table until the connector for that table supports it. That is true of change data capture, data source transactions and several INSERT extensions, and each section says which side of the line it is on.

The stack, and where each decision lives

   clients                                       servers and engines
   -------                                       -------------------
   Python (thin client)  --\
   Scala / Java          ---+--> Spark Connect --> driver --> Catalyst
   JDBC driver (4.1)     ---+      server                    (parser, analyzer,
   REST (4.3)            --/       (4.0 -> 4.3)               optimiser, planner)
                                                                  |
        declarative layer                                         v
        -----------------                              execution engine
        Spark Declarative Pipelines (4.1)              batch, micro-batch,
          Auto CDC: SCD 1 (4.2), SCD 2 (4.3)           real-time mode (4.1, 4.3)
                    |                                          |
                    v                                          v
              SQL surface                              Data Source V2
              pipe, scripting, QUALIFY,                catalogs, connectors,
              ASOF, JSON_TABLE                         transactions (4.2)
                                                       file formats, JDBC

Read the diagram from the left. A client reaches the server through Spark Connect, whichever language or protocol it uses. The declarative layer produces SQL and DataFrame work that Catalyst plans like any other. Execution is one of three modes, and every write or read goes out through Data Source V2. The numbers mark where in the line each box got its new capability, which is also the order the next section takes them in.

Under the hood: the features, release by release

Every entry has the same shape, so you can skim for the one you need: what it is and why it exists, the syntax, a run, and a maturity tag from the table above. The runs are real. The sample data is a four-row orders view (id, region, amount) unless the entry says otherwise, and the output shown is what the release printed. Where a feature was not run, the entry says so and says why.

Spark 4.0: the break, and a far more expressive SQL

All 4.0 examples ran on apache/spark:4.0.5-python3, which reports Scala 2.13 and Java 17. Those two facts are themselves the baseline change: Scala 2.12 and JDK 8 and 11 are gone, which is covered in the release history post.

ANSI mode on by default (SPARK-44444)

Maturity: Default.

Before 4.0, an invalid cast returned NULL and a divide by zero returned NULL, silently. With ANSI mode on, both fail loudly, and the error message names the function that tolerates the bad value instead. Data you used to lose without noticing now stops the job.

SET spark.sql.ansi.enabled;          -- true

SELECT CAST('abc' AS INT);           -- [CAST_INVALID_INPUT] ... Use `try_cast` to tolerate malformed input
SELECT try_cast('abc' AS INT);       -- NULL
SELECT 1/0;                          -- [DIVIDE_BY_ZERO] ... Use `try_divide` to tolerate divisor being 0
SELECT try_divide(1, 0);             -- NULL

Watch for: this is the change most likely to turn a green pipeline red. Run your own data through it before the upgrade, and use the try_* functions where a NULL was the intended answer.

VARIANT, a type for JSON you do not want to schema

Maturity: Default in 4.0, generally available in 4.1.

A VARIANT holds a semi-structured value, such as JSON, in an encoded binary form, so a query reads a path out of it without re-parsing text and without a fixed schema up front. The alternative is a STRING column and get_json_object on every read.

SELECT parse_json('{"a":1,"b":[1,2,3],"c":{"d":"x"}}') AS v;
-- {"a":1,"b":[1,2,3],"c":{"d":"x"}}

SELECT variant_get(parse_json('{"a":1,"b":[1,2,3]}'), '$.b[1]', 'int');   -- 2

SELECT schema_of_variant(parse_json('{"a":1,"b":[1,2,3]}'));
-- OBJECT<a: BIGINT, b: ARRAY<BIGINT>>

The v:field colon shorthand is not in 4.0. The same statement fails on 4.0.5 with PARSE_SYNTAX_ERROR ... at or near ':', and works from 4.1.

SQL pipe syntax (SPARK-49555)

Maturity: Default.

Pipe syntax lets a query read top to bottom, one step per |>, instead of inside-out. It is the same logical plan, written in the order you think it.

FROM orders
|> WHERE amount > 5
|> AGGREGATE sum(amount) AS total GROUP BY region
|> ORDER BY total DESC;
-- eu  40.0
-- us  20.0

FROM orders |> EXTEND amount * 2 AS doubled |> SELECT id, doubled |> LIMIT 2;
-- 1  20.0
-- 2  40.0

SQL user-defined functions (SPARK-46057)

Maturity: Default.

A function written in SQL, scalar or table-valued. Because it is SQL, the optimiser can see through it, which a Python or Scala UDF does not allow.

CREATE TEMPORARY FUNCTION add_tax(x DOUBLE) RETURNS DOUBLE RETURN x * 1.18;
SELECT id, add_tax(amount) AS gross FROM orders;        -- 1  11.799999999999999 ...

CREATE TEMPORARY FUNCTION big_orders(min_amt DOUBLE)
  RETURNS TABLE (id INT, amount DOUBLE)
  RETURN SELECT id, amount FROM orders WHERE amount >= min_amt;
SELECT * FROM big_orders(20.0);                          -- (2, 20.0), (3, 30.0)

Session variables (SPARK-42849)

Maturity: Default.

A typed variable that lives for the session, set with DECLARE and SET VAR and usable anywhere an expression is. It is the typed alternative to splicing a value into SQL text.

DECLARE OR REPLACE VARIABLE threshold DOUBLE DEFAULT 15.0;
SELECT id FROM orders WHERE amount > threshold;          -- 2, 3

SET VAR threshold = 25.0;
SELECT id FROM orders WHERE amount > threshold;          -- 3

String collation (SPARK-46830)

Maturity: Default.

A collation controls how strings compare and sort: case-insensitive, accent aware, by locale. Before collations, case-insensitive matching meant wrapping both sides in lower().

SELECT 'Apple' = 'apple';                                -- false
SELECT 'Apple' COLLATE UTF8_LCASE = 'apple';             -- true
SELECT collation('x' COLLATE UNICODE_CI);                -- SYSTEM.BUILTIN.UNICODE_CI

-- values: 'Ana', 'ana', 'ANA', 'Bob'
SELECT count(DISTINCT n COLLATE UTF8_LCASE) FROM names;  -- 2

The built-in XML data source (SPARK-44265)

Maturity: Default.

XML reads and writes without the old external package, plus from_xml, to_xml and schema_of_xml for XML held in a string column.

SELECT from_xml('<p><name>Ana</name><age>30</age></p>', 'name STRING, age INT');
-- {name: Ana, age: 30}

SELECT schema_of_xml('<p><name>Ana</name><age>30</age></p>');
-- STRUCT<age: BIGINT, name: STRING>

df = spark.read.format("xml").option("rowTag", "person").load("/tmp/xmlin")
# age: long, name: string  ->  (30, 'Ana'), (41, 'Bob')

The Python Data Source API and Python UDTFs (SPARK-44076)

Maturity: Default, but needs the pyarrow package.

You can write a data source in plain Python, register it, and read it by name, including from SQL. A Python user-defined table function returns several rows per input row. On the official image both fail until pyarrow is installed: the Python data source worker raises ModuleNotFoundError: No module named 'pyarrow'.

from pyspark.sql.datasource import DataSource, DataSourceReader
from pyspark.sql.functions import udtf

class CounterDataSource(DataSource):
    @classmethod
    def name(cls): return "counter"
    def schema(self): return "n INT, square INT"
    def reader(self, schema): return CounterReader(self.options)

class CounterReader(DataSourceReader):
    def __init__(self, options): self.limit = int(options.get("limit", 3))
    def read(self, partition):
        for i in range(1, self.limit + 1):
            yield (i, i * i)

spark.dataSource.register(CounterDataSource)
spark.read.format("counter").option("limit", 4).load().show()
# n=1..4, square=1,4,9,16

@udtf(returnType="word: string, length: int")
class SplitWords:
    def eval(self, text: str):
        for w in text.split():
            yield (w, len(w))

spark.udtf.register("split_words", SplitWords)
spark.sql("SELECT * FROM split_words('spark four point zero')").show()
# spark 5, four 4, point 5, zero 4

Also in 4.0

Spark 4.1: pipelines, procedural SQL, and real-time mode

The 4.1 examples ran on apache/spark:4.1.3-python3, apart from real-time mode, which used the playground’s Spark 4.1.3 image because it needs Kafka.

SQL scripting (SPARK-54499)

Maturity: GA, on by default (spark.sql.scripting.enabled is true).

A script is a BEGIN ... END block with variables, loops, conditions and error handlers, run as one statement. It moves logic that lived in a Python driver into SQL that an analyst can read.

BEGIN
  DECLARE total INT DEFAULT 0;
  DECLARE i INT DEFAULT 1;
  WHILE i <= 5 DO
    SET total = total + i;
    SET i = i + 1;
  END WHILE;
  SELECT total, i;
END;
-- 15, 6

BEGIN
  DECLARE result DOUBLE DEFAULT 0;
  DECLARE EXIT HANDLER FOR DIVIDE_BY_ZERO
  BEGIN
    SET result = -1;
    SELECT 'caught divide by zero' AS status, result;
  END;
  SET result = 1 / 0;
  SELECT 'not reached' AS status, result;
END;
-- caught divide by zero, -1.0

BEGIN
  DECLARE msg STRING DEFAULT '';
  FOR r AS SELECT region, sum(amount) AS total FROM orders GROUP BY region ORDER BY region DO
    SET msg = msg || r.region || '=' || CAST(r.total AS STRING) || ' ';
  END FOR;
  SELECT trim(msg) AS summary;
END;
-- eu=40.0 us=25.0

A script returns the result of its last query, which is what spark.sql handed back in each run above.

Recursive common table expressions (SPARK-24497)

Maturity: Default.

A CTE that refers to itself, for hierarchies and graph walks that used to need a loop in application code.

WITH RECURSIVE t(n) AS (
  SELECT 1
  UNION ALL
  SELECT n + 1 FROM t WHERE n < 5
) SELECT n FROM t;
-- 1, 2, 3, 4, 5

A four-level org chart walked from the root (boss IS NULL) returned ceo 0, vp 1, dir 2, mgr 3, one row per level, by joining each employee to the previous level’s output.

VARIANT generally available, with colon paths (SPARK-54454, SPARK-52494)

Maturity: GA, on by default.

4.1 declares VARIANT generally available and adds the v:path shorthand, with ::type to cast the result. The release also adds shredding for faster Parquet reads of variants; that part was not run here.

SELECT v:user::string AS user,
       v:tags[0]::string AS first_tag,
       schema_of_variant(v:n) AS n_type
FROM ev ORDER BY 1;
-- a  x     BIGINT
-- b  NULL  STRING

The n field is a number in one row and a string in the other, which a typed column could not hold, and schema_of_variant reports each row’s own type.

Sketch functions (SPARK-52515, SPARK-54199, SPARK-52407)

Maturity: Default.

Approximate aggregates that can be stored and merged: top-k frequent items, KLL quantile sketches and Theta set-cardinality sketches. A sketch column can be rolled up later without rereading the raw rows.

SELECT approx_top_k(page, 2) FROM clicks;
-- [{item: a, count: 3}, {item: b, count: 2}]

SELECT kll_sketch_get_quantile_bigint(kll_sketch_agg_bigint(ms), 0.5D) AS p50,
       kll_sketch_get_quantile_bigint(kll_sketch_agg_bigint(ms), 0.99D) AS p99
FROM lat;                      -- ms = 1..1000
-- 499, 989 on 4.1.3 (500, 990 on 4.1.4)

SELECT theta_sketch_estimate(theta_union_agg(s)) FROM (
  SELECT theta_sketch_agg(x) s FROM VALUES (1),(2),(3) AS a(x)
  UNION ALL
  SELECT theta_sketch_agg(x)   FROM VALUES (3),(4)   AS b(x));
-- 4

The exact median of 1 to 1000 is about 500 and the exact 99th percentile about

1. The sketch returned 499 and 989 on the 4.1.3 image and 500 and 990 on the 4.1.4 tarball, so expect values within a rank or two of exact that can differ between runs and patch releases.

Parameter markers wherever you need them (SPARK-53573)

Maturity: Default.

:name markers used to work in a few expression positions. In 4.1 a pre-parser substitutes them nearly anywhere, including identifiers and inside scripts, so a value from application code never has to be spliced into the SQL text.

spark.sql("SELECT * FROM IDENTIFIER(:t) ORDER BY id LIMIT :n", args={"t": "orders", "n": 2})
# (1, eu, 10.0), (2, us, 20.0)

spark.sql("""BEGIN
  DECLARE x INT DEFAULT :start;
  SET x = x + 1;
  SELECT x;
END""", args={"start": 41})
# 42

Arrow-native UDFs (SPARK-52214, SPARK-52979)

Maturity: Default, needs pyarrow.

@arrow_udf and @arrow_udtf pass a function whole PyArrow arrays instead of converting to pandas first, so a vectorised function stays vectorised.

import pyarrow as pa, pyarrow.compute as pc
from pyspark.sql.functions import arrow_udf, col

@arrow_udf("double")
def with_tax(amount: pa.Array) -> pa.Array:
    return pc.multiply(amount, 1.18)

df.select("id", with_tax(col("amount")).alias("gross")).show()
# 1  11.799999999999999
# 2  23.599999999999998

Spark Declarative Pipelines (SPARK-51727)

Maturity: Default, driven by the spark-pipelines CLI. The official image needs extra Python packages.

You declare datasets, and Spark works out the order. A pipeline project is a spark-pipeline.yml and a folder of Python or SQL files. This one has a materialized view in Python and another in SQL that reads it:

name: orders_pipeline
libraries:
  - glob:
      include: transformations/**
storage: file:///work/sdp/storage

# transformations/orders.py
from pyspark import pipelines as dp
from pyspark.sql import DataFrame, SparkSession
spark = SparkSession.active()

@dp.materialized_view
def orders() -> DataFrame:
    return spark.sql("SELECT * FROM VALUES (1,'eu',10.0D),(2,'us',20.0D),"
                     "(3,'eu',30.0D),(4,'us',5.0D) AS t(id, region, amount)")

-- transformations/revenue.sql
CREATE MATERIALIZED VIEW revenue_by_region AS
SELECT region, sum(amount) AS total, count(*) AS orders
FROM orders
GROUP BY region;

spark-pipelines dry-run      # catches syntax, analysis and cycle errors without writing
spark-pipelines run

The run log shows both flows queued together and orders running first, because revenue_by_region reads it. Nothing in the files states that order. The output tables were:

orders:                          revenue_by_region:
 id region amount                 region total orders
  1 eu     10.0                   eu      40.0  2
  2 us     20.0                   us      25.0  2
  3 eu     30.0
  4 us      5.0

The official image does not contain the Python packages the CLI needs. The first run fails with ModuleNotFoundError: No module named 'yaml', and the CLI works through Spark Connect, so grpcio and its companions are needed too. Install pyyaml grpcio grpcio-status protobuf googleapis-common-protos zstandard, or use pip install pyspark[pipelines], which the 4.3 documentation recommends.

Structured Streaming real-time mode (SPARK-53736)

Maturity: Default for stateless queries, Scala only, Kafka source.

Real-time mode runs a streaming query as long-lived tasks that process records as they arrive, instead of cutting the stream into micro-batches. You choose it with a trigger. In 4.1 it supports stateless queries and a short allowlist of sources, sinks and operators; the Kafka source is the supported source, and console, memory, foreach and Kafka are the supported sinks.

import org.apache.spark.sql.streaming.Trigger
import org.apache.spark.sql.functions._

val out = spark.readStream.format("kafka")
  .option("kafka.bootstrap.servers", "kafka:29092")
  .option("subscribe", "rtm_demo_in2").option("startingOffsets", "latest").load()
  .select(upper(col("value").cast("string")).alias("v"))

val q = out.writeStream.format("console").outputMode("update")
  .trigger(Trigger.RealTime("5 seconds")).start()

Messages hello, real and time produced to the topic came out as HELLO, REAL and TIME in a single batch. Python does not accept the trigger on 4.1.3: trigger(realTime="5 seconds") raised TypeError: DataStreamWriter.trigger() got an unexpected keyword argument 'realTime', matching the release note’s “Scala stateless support”. Stateful queries wait for 4.3.

Spark 4.2: new types, new joins, and a clearer SQL surface

The 4.2 examples ran on apache/spark:4.2.1-python3. Three of the release’s headline items, change data capture, INSERT ... WITH SCHEMA EVOLUTION and Data Source V2 transactions, are defined by the engine but implemented by a connector, so the built-in sources cannot show them. They are in the table at the end of this section.

Geospatial types: GEOMETRY and GEOGRAPHY (SPARK-51658)

Maturity: Default in 4.2. The same call on 4.1.3 fails with GEOSPATIAL_DISABLED.

Two first-class types for spatial values, each carrying an SRID (spatial reference id), so a geometry knows its coordinate system instead of being an anonymous BINARY. The release notes mention ST_* functions, WKB and WKT, and Parquet I/O. On 4.2.1 the function surface is small: SHOW FUNCTIONS lists five, and 4.3.0 lists the same five.

-- the five: st_asbinary, st_geogfromwkb, st_geomfromwkb, st_setsrid, st_srid
SELECT typeof(ST_GeomFromWKB(X'0101000000000000000000F03F0000000000000040'));   -- geometry(0)
SELECT typeof(ST_GeogFromWKB(X'0101000000000000000000F03F0000000000000040'));   -- geography(4326)

SELECT ST_SRID(ST_SetSRID(ST_GeomFromWKB(X'0101000000000000000000F03F0000000000000040'), 4326));
-- 4326

CREATE TEMP VIEW places AS
SELECT 'a' AS name,
       ST_SetSRID(ST_GeomFromWKB(X'0101000000000000000000F03F0000000000000040'), 4326) AS g;
DESCRIBE places;        -- name string, g geometry(4326)

Watch for: a GEOMETRY defaults to SRID 0 (unknown), a GEOGRAPHY to 4326, and there are no distance or containment functions in this release, so the types are a storage and interchange foundation rather than a spatial query engine yet.

The QUALIFY clause (SPARK-31561)

Maturity: Default.

QUALIFY filters on a window function directly, so “the top row per group” no longer needs a subquery around the window.

SELECT id, region, amount FROM orders_t
QUALIFY row_number() OVER (PARTITION BY region ORDER BY amount DESC, id) = 1
ORDER BY region;
-- 3  eu  30.0
-- 2  us  20.0

SELECT id, region, rank() OVER (PARTITION BY region ORDER BY amount DESC) AS r
FROM orders_t QUALIFY r = 1 ORDER BY id;     -- aliases work; ties all qualify
-- 2 us 1,  3 eu 1,  5 eu 1

The condition must contain a window function, and aggregates are not allowed in it.

NEAREST BY top-K ranking join (SPARK-56395)

Maturity: Default. Evaluation is brute force today.

For each row on the left, return the K best rows on the right under a ranking expression. It is the join shape behind “closest store”, “most similar item” and fuzzy matching, written as one clause instead of a cross join and a window.

-- pts:  (q1, 1.0), (q2, 9.0)       base: (a, 0.0), (b, 2.0), (c, 8.0), (d, 20.0)
SELECT q, x, name, y
FROM pts JOIN base EXACT NEAREST 2 BY DISTANCE abs(x - y)
ORDER BY q, abs(x - y);
-- q1 1.0 a 0.0
-- q1 1.0 b 2.0
-- q2 9.0 c 8.0
-- q2 9.0 b 2.0

SELECT q, name FROM pts JOIN base EXACT NEAREST BY DISTANCE abs(x - y);   -- K defaults to 1
-- q1 a,  q2 c

Syntax: { APPROX | EXACT } NEAREST [ k ] BY { DISTANCE | SIMILARITY } expr. INNER and LEFT OUTER are the supported join types, and k is between 1 and 100000. APPROX lets the optimiser use a faster strategy when one exists. The 4.2.1 documentation states the cost plainly: the current implementation evaluates the full cross product of the two sides, so pre-filter a large right side.

SET PATH: where unqualified names are looked up (SPARK-54806)

Maturity: Behind a flag (spark.sql.path.enabled, default false).

The SQL path is an ordered list of schemas Spark searches to resolve an unqualified table, view, function or variable. SET PATH changes it for the session, and current_path() reads it. Persistent views and SQL functions capture the path when they are created.

SELECT current_path();    -- system.builtin,system.session,spark_catalog.default

SET PATH = DEFAULT_PATH;
-- [UNSUPPORTED_FEATURE.SET_PATH_WHEN_DISABLED] SET PATH is disabled.
-- Set spark.sql.path.enabled to true to enable it.

SET spark.sql.path.enabled = true;
-- two schemas each holding a table named facts: ref.facts (2,'ref'), sales.facts (1,'sales')

SET PATH = spark_catalog.ref, DEFAULT_PATH;
SELECT current_path();    -- spark_catalog.ref,system.builtin,system.session,spark_catalog.default
SELECT * FROM facts;      -- (2, 'ref')

SET PATH = spark_catalog.sales, DEFAULT_PATH;
SELECT * FROM facts;      -- (1, 'sales')

DDL does not use the path: CREATE TABLE t always creates t in the current schema.

Metric views (SPARK-54119)

Maturity: Default, on the built-in catalog.

A metric view defines dimensions and measures once, in YAML, so every consumer computes the same number. You query a measure with MEASURE(...) and group by a dimension.

CREATE OR REPLACE VIEW sales.revenue_mv WITH METRICS LANGUAGE YAML AS $$
version: "0.1"
source: spark_catalog.sales.orders
dimensions:
  - name: region
    expr: region
measures:
  - name: total_amount
    expr: sum(amount)
  - name: order_count
    expr: count(1)
$$;

SELECT region, MEASURE(total_amount) AS total, MEASURE(order_count) AS n
FROM sales.revenue_mv GROUP BY region ORDER BY region;
-- eu  70.0  3
-- us  25.0  2

The 4.2.1 documentation has no page for this syntax. The statement form comes from the project’s own test suite, and the YAML layout above is the one that ran.

time_bucket and IGNORE NULLS (SPARK-56594, SPARK-55256)

Maturity: Default.

time_bucket(size, ts[, origin]) returns the start of the bucket a timestamp falls into, replacing the date_trunc or epoch-division arithmetic everyone writes for fixed-width windows. collect_list and array_agg now take IGNORE NULLS and RESPECT NULLS.

SELECT time_bucket(INTERVAL '15' MINUTE, TIMESTAMP_NTZ '2026-11-01 11:27:00');
-- 2026-11-01 11:15:00

SELECT time_bucket(INTERVAL '1' HOUR, ts) AS hour, count(*) AS n, sum(v) AS total
FROM (VALUES (TIMESTAMP_NTZ '2026-11-01 10:05:00', 1),
             (TIMESTAMP_NTZ '2026-11-01 10:40:00', 2),
             (TIMESTAMP_NTZ '2026-11-01 11:10:00', 5)) AS t(ts, v)
GROUP BY 1 ORDER BY 1;
-- 10:00  2  3
-- 11:00  1  5

SELECT collect_list(x) IGNORE NULLS, collect_list(x) RESPECT NULLS
FROM VALUES (1), (CAST(NULL AS INT)), (3) AS t(x);
-- [1, 3]    [1, NULL, 3]

Also in 4.2

Spark 4.3: stateful real-time mode, semi-structured SQL, and history tracking

All 4.3 examples ran on apache/spark:4.3.0-python3, which reports Scala 2.13 and Java 21. The release shipped without preview builds, and several features below are behind a flag or labelled preview.

JSON_TABLE (SPARK-58366)

Maturity: Default. Flat form only.

The SQL-standard way to turn a JSON document into rows and typed columns. It is the construct that migrated Oracle, DB2 and MySQL queries expect.

SELECT t.* FROM JSON_TABLE(
  '{"items":[{"id":1,"n":"a"},{"id":2,"n":"b"}]}',
  '$.items[*]'
  COLUMNS (
    seq  FOR ORDINALITY,
    id   INT    PATH '$.id',
    name STRING PATH '$.n'
  )
) AS t;
-- 1  1  a
-- 2  2  b

SELECT * FROM JSON_TABLE(
  '{"rows":[{"id":10,"opt":1},{"id":20}]}',
  '$.rows[*]'
  COLUMNS (id INT, hasOpt BOOLEAN EXISTS PATH '$.opt')
) AS t;
-- 10  true
-- 20  false

-- against a column, with LATERAL
SELECT d.id, t.k, t.v FROM docs d,
LATERAL JSON_TABLE(d.doc, '$.tags[*]' COLUMNS (k STRING PATH '$.k', v INT PATH '$.v')) AS t;
-- 1 x 1,  1 y 2,  2 z 3

NESTED PATH columns are not supported yet, so a document nested more than one array deep still needs two steps.

ASOF JOIN (SPARK-58092)

Maturity: Behind a flag (spark.sql.join.asofJoin.enabled, default false).

For each left row, take the closest preceding row on the right: the quote in force when a trade happened, the price list valid on an order date. Without it you write a window function plus a join.

SELECT t.id, t.sym, t.ts, q.ts AS quote_ts, q.bid
FROM trades t ASOF JOIN quotes q
  MATCH_CONDITION (t.ts >= q.ts)
  ON t.sym = q.sym
ORDER BY t.id;

With the flag off the statement fails at parse time with [UNSUPPORTED_FEATURE.ASOF_JOIN] ... Set "spark.sql.join.asofJoin.enabled" to true. With it on, using trades at 10:00:05 and 10:00:20 for AAPL and 10:00:07 for MSFT, against quotes at 10:00:00 and 10:00:10 for AAPL and 10:00:06 for MSFT:

id  sym   ts        quote_ts  bid
1   AAPL  10:00:05  10:00:00  100.0
2   AAPL  10:00:20  10:00:10  101.0
3   MSFT  10:00:07  10:00:06  200.0

Only INNER and LEFT OUTER are supported. In streaming, micro-batch supports a streaming left side joined to a static right side, and real-time mode does not support it.

UNNEST in the FROM clause (SPARK-58419)

Maturity: Default.

The ANSI way to expand arrays into rows, in place of explode and LATERAL VIEW. Several arrays expand in parallel and the shorter ones pad with NULL.

SELECT * FROM UNNEST(array('a','b')) WITH ORDINALITY AS t(value, pos);
-- a 1,  b 2

SELECT * FROM UNNEST(array(1,2), array(10,20,30)) AS t(a, b);
-- 1 10,  2 20,  NULL 30

SELECT id, elem
FROM VALUES (1, array(10,20)), (2, array(30)) AS data(id, arr),
     LATERAL UNNEST(arr) AS t(elem);
-- 1 10,  1 20,  2 30

VARIANT editing functions (SPARK-57804 and others)

Maturity: Default.

Until now a variant could be read but not changed without rebuilding it. These return a new variant with a path set, inserted, deleted or appended to.

SELECT variant_set(parse_json('{"a":1,"tags":["x"]}'), '$.b', 3);
-- {"a":1,"b":3,"tags":["x"]}

SELECT variant_insert(parse_json('["a","b","c"]'), '$[1]', 'z');
-- ["a","z","b","c"]

SELECT variant_delete(parse_json('{"a":1,"b":2,"items":[1,2,3]}'), '$.a', '$.items[0]');
-- {"b":2,"items":[2,3]}

SELECT variant_array_append(parse_json('{"a":[1,2]}'), '$.a', 3);
-- {"a":[1,2,3]}

SELECT variant_strip_nulls(parse_json('{"a":1,"b":null,"c":[1,null]}'));
-- {"a":1,"c":[1]}

variant_set and variant_insert differ on an existing key: set replaces it, insert is an error. try_ forms exist too (try_variant_set and others).

TIME(p) and nanosecond timestamps (SPARK-57551, SPARK-56876)

Maturity: Behind flags, and preview for the nanosecond timestamps.

TIME(p) goes up to nanosecond precision, and TIMESTAMP_NTZ(9) and TIMESTAMP_LTZ(9) add nanosecond timestamps. Two flags guard them: spark.sql.timeType.enabled for TIME, and spark.sql.timestampNanosTypes.enabled for the timestamps, which the release notes call a preview.

SET spark.sql.timeType.enabled = true;
SET spark.sql.timestampNanosTypes.enabled = true;

SELECT typeof(CAST('10:15:30.123456789' AS TIME(9)));        -- time(9)

SELECT CAST(CAST('2026-11-01 10:00:00.123456789' AS TIMESTAMP_NTZ(9)) AS STRING),
       typeof(CAST('2026-11-01 10:00:00.123456789' AS TIMESTAMP_NTZ(9)));
-- 2026-11-01 10:00:00.123456789   timestamp_ntz(9)

SELECT CAST('2026-11-01 10:00:00.123456789' AS TIMESTAMP_NTZ);   -- default keeps microseconds
-- 2026-11-01 10:00:00.123456

Without spark.sql.timeType.enabled, CAST(... AS TIME(9)) fails with UNSUPPORTED_TIME_TYPE, so TIME itself is still behind a flag in 4.3.0. The Python client in this release cannot materialise a TIMESTAMP_NTZ(9) result: selecting the column raised CANNOT_PARSE_DATATYPE ... timestamp_ntz(9), which is why the example casts to a string.

Stateful real-time mode (SPARK-58635, SPARK-57228, SPARK-58508)

Maturity: Default for the supported operators; needs one task slot per shuffle partition.

4.1 gave real-time mode to stateless queries. 4.3 adds streaming aggregation, dropDuplicates and transformWithState. The trigger is the same.

val counts = spark.readStream.format("kafka")
  .option("kafka.bootstrap.servers", "kafka:29092").option("subscribe", "rtm43_demo3")
  .option("startingOffsets", "latest").load()
  .select(col("value").cast("string").alias("word"))
  .groupBy("word").count()

counts.writeStream.format("console").outputMode("update")
  .option("checkpointLocation", "/tmp/rtm43_ckpt3")
  .trigger(Trigger.RealTime("5 seconds")).start()

Producing a, b, a, a printed the running counts as the records arrived:

Batch 2
|word|count|
|   a|    1|
|   a|    2|
|   a|    3|
|   b|    1|

The word a appears three times in one batch because real-time mode emits each update as it happens, not one row per key at the end of the batch.

Watch for: the first attempt, with Spark’s default 200 shuffle partitions on a local[2] session, failed:

[CONCURRENT_SCHEDULER_INSUFFICIENT_SLOT] Cannot run the pipelined stage group:
it needs 201 concurrent task slots to run all its stages together, but only 2
are currently free.

The stages of a real-time query run at the same time, so the cluster needs a free slot for every task in all of them. Set spark.sql.shuffle.partitions to a number the cluster can run concurrently (2, on a local[6] session, worked) before you size the job.

Auto CDC with SCD Type 2 (SPARK-58321, SPARK-58247, SPARK-58261)

Maturity: Needs a connector. Spark’s built-in file formats are refused.

Auto CDC applies a stream of change events to a target table inside a Declarative Pipeline. Type 1 keeps only the latest row per key. Type 2, new in 4.3, keeps the history of every change. The SQL form is a streaming table with an Auto CDC flow:

CREATE STREAMING TABLE customers_history
FLOW AUTO CDC
FROM STREAM(cdc_events)
KEYS (id)
APPLY AS DELETE WHEN op = 'DELETE'
SEQUENCE BY version
COLUMNS * EXCEPT (op)
STORED AS SCD TYPE 2;

On a pipeline whose target is a default Parquet table, the syntax parses and the flow fails to start:

[AUTOCDC_TARGET_DOES_NOT_SUPPORT_MERGE] Cannot start AutoCDC flow: the target table
`spark_catalog`.`default`.`customers_history` (format: parquet) does not support
row-level operations.

Each micro-batch is applied with a MERGE, so the target must be a table whose connector implements Data Source V2 row-level operations. The 4.3 documentation names lakehouse connectors such as Apache Iceberg as the usual choice, and it notes that its own page for SCD Type 2 is still to be written. This is the clearest case in the line of an engine feature that does nothing until a connector supports it. A pipeline run against an Iceberg catalog was not done here.

New functions (SPARK-56820, SPARK-57253, SPARK-58399, SPARK-58295)

Maturity: Default.

-- counter_diff: turn a cumulative counter into per-interval deltas
SELECT t, c, counter_diff(c) OVER (ORDER BY t) AS delta FROM VALUES
  (TIMESTAMP_NTZ '2026-01-01 00:00:00', 100),
  (TIMESTAMP_NTZ '2026-01-01 00:01:00', 130),
  (TIMESTAMP_NTZ '2026-01-01 00:02:00', 190) AS m(t, c);
-- 00:00  100  NULL
-- 00:01  130  30
-- 00:02  190  60

SELECT collect_union(col) FROM VALUES (array(1,2)), (array(2,3)), (array(1)) AS tab(col);
-- [1, 2, 3]

SELECT jaro_winkler_similarity('MARTHA', 'MARHTA');              -- 0.9611111111111111
SELECT to_base32(CAST('spark' AS BINARY)),
       CAST(from_base32(to_base32(CAST('spark' AS BINARY))) AS STRING);
-- ONYGC4TL   spark

Faster queries on small data (SPARK-57851, SPARK-57268)

Maturity: Behind flags. Not demonstrated here.

Two opt-in settings aim at laptop-sized data. spark.sql.optimizer.singleTaskExecution.enabled (default false) lets eligible fragments that read a small single-partition scan run in one task and skip the shuffle before a sort or aggregation. spark.sql.execution.arrow.cache.prefetch.enabled (default false) prefetches and decompresses the next batch of a compressed Arrow cache in a background thread. A trivial single-partition aggregate showed no plan difference with the first flag on, since it needed no exchange either way, so there is no number to quote. Measure it on your own small queries.

Also in 4.3

Hands-on demo: run every example yourself

Everything in the previous section ran on one of two setups: an official apache/spark Docker image, and the plain release tarball from Apache with nothing else installed. This section gives you the second one, because it needs only Java and shows what a clean install does.

What you need

These are the latest releases of each line at the time of writing. The Python column comes from each tarball’s own Requires-Python, and all four ran on the same Java 17 (Temurin 17.0.20), including 4.3.0.

The Python column decides which examples you can run from a tarball. A machine with Python 3.9 can run the SQL and Scala examples on all four releases and the PySpark examples on 4.0 only.

Download and verify

The tarball comes from the Apache CDN and the checksum from the main Apache distribution site, so the two come from different hosts. The CDN keeps only current releases; an older patch release is under https://archive.apache.org/dist/spark/. This is the script that fetched the four files above, each one verified:

mkdir -p ~/spark-4.x && cd ~/spark-4.x
for v in 4.0.5 4.1.4 4.2.1 4.3.0; do
  f=spark-$v-bin-hadoop3.tgz
  curl -sSf -C - -O "https://dlcdn.apache.org/spark/spark-$v/$f"
  curl -sSf -o "$f.sha512" "https://downloads.apache.org/spark/spark-$v/$f.sha512"
  got=$(shasum -a 512 "$f" | awk '{print $1}')
  if tr -d ' \n\r' < "$f.sha512" | tr 'A-F' 'a-f' | grep -q "$got"; then
    echo "OK   $f"; tar -xzf "$f"
  else
    echo "BAD  $f"
  fi
done

Four things that went wrong first

Each of these cost a run, so they are worth knowing before you start.

A different Spark answers. The first run printed SPARK 2.4.4. The shell had SPARK_HOME pointing at an older install, and the spark-shell script honours it over the directory it lives in. Every result from that run was invalid. Start Spark with a clean environment (env -i) and print spark.version first.

spark-shell -i script ran nothing. On this setup the REPL started and quit without executing the file. Feeding the script on standard input (< script.scala) worked on all four releases.

The driver address. Without pinning the driver to localhost, a run hung and logged RpcEndpointNotFoundException: Cannot find endpoint: spark://CoarseGrainedScheduler@.... With spark.driver.bindAddress and spark.driver.host set to 127.0.0.1 every run completed.

Python is too old. The 4.1 and 4.2 tarballs declare Python 3.10 or newer and 4.3 declares 3.11. On 4.2.1 with Python 3.9 a Python worker failed with a TypeError, so a stock macOS Python runs only the 4.0 PySpark examples.

A runner that executes a SQL file

The runner reads a file of statements separated by a line containing only ;;. A line starting with -- ### is printed as a heading, and the SQL after it runs. It uses only spark.sql, so any statement from the previous section works in it.

import scala.io.Source

val specPath = sys.env("SPEC")
val statements = Source.fromFile(specPath).mkString.split("\n;;\n").map(_.trim).filter(_.nonEmpty)
println("SPARK " + spark.version + " | Scala " + scala.util.Properties.versionNumberString + " | Java " + System.getProperty("java.version"))

statements.foreach { block =>
  val lines = block.linesIterator.toList
  // A leading "-- ### title" line is printed as a heading; whatever follows it is the SQL.
  val (heading, sql) =
    if (lines.head.startsWith("-- ###")) (Some(lines.head.drop(7)), lines.tail.mkString("\n").trim)
    else (None, block)
  heading.foreach(h => println("\n### " + h))
  if (sql.nonEmpty) {
    println(sql.linesIterator.next().take(110))
    try spark.sql(sql).collect().take(12).foreach(r => println("   " + r.mkString(" | ")))
    catch { case e: Throwable => println("   ERROR: " + e.getMessage.linesIterator.next().take(200)) }
  }
}
sys.exit(0)

Save it as runner.scala, and run any SQL file with a function like this, where JAVA is a JDK 17 home:

run() {   # run <release> <sql-file>
  env -i HOME="$HOME" PATH=/usr/bin:/bin JAVA_HOME="$JAVA" SPEC="$2" \
    ~/spark-4.x/spark-$1-bin-hadoop3/bin/spark-shell --master 'local[2]' \
    --conf spark.ui.enabled=false --conf spark.eventLog.enabled=false \
    --conf spark.driver.bindAddress=127.0.0.1 --conf spark.driver.host=127.0.0.1 \
    --conf spark.sql.shuffle.partitions=2 < runner.scala
}
run 4.0.5 smoke40.sql

One SQL file per release

Each file runs the headline features of its release from the previous section. They are the four files that were run, one per tarball, and the results listed under each are what the run printed.

smoke40.sql, on 4.0.5

-- ### setup
CREATE OR REPLACE TEMP VIEW orders AS SELECT * FROM VALUES (1,'eu',10.0D),(2,'us',20.0D),(3,'eu',30.0D),(4,'us',5.0D) AS t(id,region,amount)
;;
-- ### ANSI mode: try_cast tolerates a bad value (a plain CAST would raise CAST_INVALID_INPUT)
SELECT try_cast('abc' AS INT)
;;
-- ### VARIANT
SELECT variant_get(parse_json('{"a":1,"b":[1,2,3]}'), '$.b[1]', 'int')
;;
-- ### pipe syntax
FROM orders |> WHERE amount > 5 |> AGGREGATE sum(amount) AS total GROUP BY region |> ORDER BY total DESC
;;
-- ### SQL user-defined function
CREATE OR REPLACE TEMPORARY FUNCTION add_tax(x DOUBLE) RETURNS DOUBLE RETURN x * 1.18
;;
SELECT id, add_tax(amount) AS gross FROM orders ORDER BY id
;;
-- ### session variable
DECLARE OR REPLACE VARIABLE threshold DOUBLE DEFAULT 15.0
;;
SELECT id FROM orders WHERE amount > threshold ORDER BY id
;;
-- ### string collation
SELECT 'Apple' COLLATE UTF8_LCASE = 'apple'
;;
-- ### XML
SELECT from_xml('<p><name>Ana</name><age>30</age></p>', 'name STRING, age INT')

try_cast           NULL
variant_get        2
pipe syntax        eu 40.0, us 20.0
SQL UDF            1 11.799999999999999 ... 4 5.8999999999999995
session variable   2, 3
collation          true
from_xml           [Ana,30]

smoke41.sql, on 4.1.4

-- ### setup
CREATE OR REPLACE TEMP VIEW clicks AS SELECT * FROM VALUES ('a'),('b'),('a'),('c'),('a'),('b'),('d') AS t(page)
;;
CREATE OR REPLACE TEMP VIEW ev AS SELECT parse_json(j) v FROM VALUES ('{"user":"a","n":3,"tags":["x","y"]}'),('{"user":"b","n":"x"}') AS t(j)
;;
-- ### SQL scripting
BEGIN
  DECLARE total INT DEFAULT 0;
  DECLARE i INT DEFAULT 1;
  WHILE i <= 5 DO
    SET total = total + i;
    SET i = i + 1;
  END WHILE;
  SELECT total, i;
END
;;
-- ### recursive CTE
WITH RECURSIVE t(n) AS (SELECT 1 UNION ALL SELECT n + 1 FROM t WHERE n < 5) SELECT n FROM t
;;
-- ### VARIANT colon path
SELECT v:user::string AS user, v:tags[0]::string AS first_tag, schema_of_variant(v:n) AS n_type FROM ev ORDER BY 1
;;
-- ### sketch functions
SELECT approx_top_k(page, 2) FROM clicks

scripting          15 | 6
recursive CTE      1, 2, 3, 4, 5
colon path         a x BIGINT   /   b NULL STRING
approx_top_k       [a,3], [b,2]

smoke42.sql, on 4.2.1

-- ### setup
CREATE OR REPLACE TEMP VIEW orders AS SELECT * FROM VALUES (1,'eu',10.0D),(2,'us',20.0D),(3,'eu',30.0D),(4,'us',5.0D),(5,'eu',30.0D) AS t(id,region,amount)
;;
CREATE OR REPLACE TEMP VIEW pts AS SELECT * FROM VALUES ('q1',1.0D),('q2',9.0D) AS t(q,x)
;;
CREATE OR REPLACE TEMP VIEW base AS SELECT * FROM VALUES ('a',0.0D),('b',2.0D),('c',8.0D),('d',20.0D) AS t(name,y)
;;
-- ### geospatial
SELECT typeof(ST_GeomFromWKB(X'0101000000000000000000F03F0000000000000040'))
;;
-- ### QUALIFY
SELECT id, region, amount FROM orders QUALIFY row_number() OVER (PARTITION BY region ORDER BY amount DESC, id) = 1 ORDER BY region
;;
-- ### NEAREST BY
SELECT q, name FROM pts JOIN base EXACT NEAREST BY DISTANCE abs(x - y) ORDER BY q
;;
-- ### SET PATH (behind a flag)
SET spark.sql.path.enabled=true
;;
SET PATH = DEFAULT_PATH
;;
SELECT current_path()
;;
-- ### time_bucket
SELECT time_bucket(INTERVAL '15' MINUTE, TIMESTAMP_NTZ '2026-11-01 11:27:00')
;;
-- ### metric view
CREATE SCHEMA IF NOT EXISTS sales
;;
CREATE TABLE sales.orders USING parquet AS SELECT * FROM orders
;;
CREATE OR REPLACE VIEW sales.revenue_mv WITH METRICS LANGUAGE YAML AS $$
version: "0.1"
source: spark_catalog.sales.orders
dimensions:
  - name: region
    expr: region
measures:
  - name: total_amount
    expr: sum(amount)
$$
;;
SELECT region, MEASURE(total_amount) AS total FROM sales.revenue_mv GROUP BY region ORDER BY region

geospatial         geometry(0)
QUALIFY            3 eu 30.0   /   2 us 20.0
NEAREST BY         q1 a   /   q2 c
SET PATH           current_path() = system.builtin,system.session,spark_catalog.default
time_bucket        2026-11-01T11:15
metric view        eu 70.0   /   us 25.0

smoke43.sql, on 4.3.0

-- ### JSON_TABLE
SELECT t.* FROM JSON_TABLE('{"items":[{"id":1,"n":"a"},{"id":2,"n":"b"}]}', '$.items[*]' COLUMNS (seq FOR ORDINALITY, id INT PATH '$.id', name STRING PATH '$.n')) AS t
;;
-- ### UNNEST
SELECT * FROM UNNEST(array(1,2), array(10,20,30)) AS t(a, b)
;;
-- ### VARIANT editing
SELECT variant_set(parse_json('{"a":1,"tags":["x"]}'), '$.b', 3)
;;
-- ### ASOF JOIN (behind a flag)
CREATE OR REPLACE TEMP VIEW trades AS SELECT * FROM VALUES (1,'AAPL',TIMESTAMP_NTZ '2026-11-01 10:00:05'),(2,'AAPL',TIMESTAMP_NTZ '2026-11-01 10:00:20') AS t(id,sym,ts)
;;
CREATE OR REPLACE TEMP VIEW quotes AS SELECT * FROM VALUES ('AAPL',TIMESTAMP_NTZ '2026-11-01 10:00:00',100.0D),('AAPL',TIMESTAMP_NTZ '2026-11-01 10:00:10',101.0D) AS q(sym,ts,bid)
;;
SET spark.sql.join.asofJoin.enabled=true
;;
SELECT t.id, q.ts AS quote_ts, q.bid FROM trades t ASOF JOIN quotes q MATCH_CONDITION (t.ts >= q.ts) ON t.sym = q.sym ORDER BY t.id
;;
-- ### TIME(9) (behind a flag)
SET spark.sql.timeType.enabled=true
;;
SELECT typeof(CAST('10:15:30.123456789' AS TIME(9)))

JSON_TABLE         1 1 a   /   2 2 b
UNNEST             1 10   /   2 20   /   NULL 30
variant_set        {"a":1,"b":3,"tags":["x"]}
ASOF JOIN          1 10:00:00 100.0   /   2 10:00:10 101.0
TIME(9)            time(9)

Parameter markers need bound arguments, so they go through the Scala API rather than a SQL file. This ran on the 4.1.4 tarball:

spark.sql("CREATE OR REPLACE TEMP VIEW orders AS SELECT * FROM VALUES (1,'eu',10.0D),(2,'us',20.0D),(3,'eu',30.0D),(4,'us',5.0D) AS t(id,region,amount)")
println("P1 " + spark.sql("SELECT * FROM orders WHERE region = :r ORDER BY id", Map("r" -> "eu")).collect().mkString(" "))
println("P2 " + spark.sql("SELECT * FROM IDENTIFIER(:t) ORDER BY id LIMIT :n", Map("t" -> "orders", "n" -> 2)).collect().mkString(" "))
println("P3 " + spark.sql("BEGIN DECLARE x INT DEFAULT :start; SET x = x + 1; SELECT x; END", Map("start" -> 41)).collect().mkString(" "))
sys.exit(0)

P1 [1,eu,10.0] [3,eu,30.0]
P2 [1,eu,10.0] [2,us,20.0]
P3 [42]

The PySpark examples

Python data source and UDTF, on the 4.0.5 tarball with Python 3.9. Create a virtual environment with pyarrow and pandas, point PYTHONPATH at the tarball’s python folder and its py4j zip, and run the script. It ran with pyarrow 21.0.0 and pandas 2.3.3.

python3 -m venv venv && venv/bin/pip install pyarrow pandas
SH=~/spark-4.x/spark-4.0.5-bin-hadoop3
env -i HOME="$HOME" PATH=/usr/bin:/bin JAVA_HOME="$JAVA" SPARK_HOME="$SH" \
  PYTHONPATH="$SH/python:$(ls $SH/python/lib/py4j-*.zip)" SPARK_LOCAL_IP=127.0.0.1 \
  PYSPARK_PYTHON=venv/bin/python venv/bin/python datasource_udtf.py

from pyspark.sql import SparkSession
from pyspark.sql.datasource import DataSource, DataSourceReader
from pyspark.sql.functions import udtf

spark = SparkSession.builder.appName("datasource-udtf").getOrCreate()
spark.sparkContext.setLogLevel("ERROR")


class CounterDataSource(DataSource):
    @classmethod
    def name(cls):
        return "counter"

    def schema(self):
        return "n INT, square INT"

    def reader(self, schema):
        return CounterReader(self.options)


class CounterReader(DataSourceReader):
    def __init__(self, options):
        self.limit = int(options.get("limit", 3))

    def read(self, partition):
        for i in range(1, self.limit + 1):
            yield (i, i * i)


spark.dataSource.register(CounterDataSource)
spark.read.format("counter").option("limit", 4).load().show()

spark.sql("CREATE OR REPLACE TEMP VIEW c USING counter OPTIONS (limit '3')")
spark.sql("SELECT n, square FROM c WHERE n > 1").show()


@udtf(returnType="word: string, length: int")
class SplitWords:
    def eval(self, text: str):
        for w in text.split():
            yield (w, len(w))


spark.udtf.register("split_words", SplitWords)
spark.sql("SELECT * FROM split_words('spark four point zero')").show()

The output was the four counter rows, the two rows with n > 1, and the four words from split_words, as shown in the previous section.

Arrow UDF, on the official 4.1.3 image. It needs Python 3.10 or newer, which the image has, and pyarrow, which it does not. Install pyarrow into a folder you mount, rather than into the image:

mkdir pylib
docker run --rm -v "$PWD":/work -w /work apache/spark:4.1.3-python3 \
  pip install --target /work/pylib pyarrow pandas
docker run --rm -v "$PWD":/work -w /work -e PYTHONPATH=/work/pylib \
  apache/spark:4.1.3-python3 /opt/spark/bin/spark-submit --master 'local[2]' arrow_udf.py

import pyarrow as pa
import pyarrow.compute as pc
from pyspark.sql import SparkSession
from pyspark.sql.functions import arrow_udf, col

spark = SparkSession.builder.appName("arrow-udf").getOrCreate()
spark.sparkContext.setLogLevel("ERROR")


@arrow_udf("double")
def with_tax(amount: pa.Array) -> pa.Array:
    return pc.multiply(amount, 1.18)


df = spark.sql("SELECT * FROM VALUES (1,'eu',10.0D),(2,'us',20.0D) AS t(id, region, amount)")
df.select("id", with_tax(col("amount")).alias("gross")).show()

Declarative Pipelines, on the official 4.1.3 image. The three project files are in the previous section. The CLI needs pyyaml, grpcio, grpcio-status, protobuf, googleapis-common-protos and zstandard as well. Install them into pylib the same way, then, from the project folder:

docker run --rm -v "$PWD/..":/work -w /work/sdp -e PYTHONPATH=/work/pylib \
  apache/spark:4.1.3-python3 /opt/spark/bin/spark-pipelines run

The run log shows orders planned and run before revenue_by_region.

Real-time mode against Kafka

Both scripts read from a Kafka broker on localhost:9092. The datalake-playground publishes that port when started with the all profile, which is what these runs used:

git clone https://github.com/rangareddy/datalake-playground.git
cd datalake-playground
PROFILE=all SPARK_VERSION=3.5.9 sh run_datalake.sh start
sh run_datalake.sh status

The Kafka source is not in the tarball, so add it with --packages matching the release, for example --packages org.apache.spark:spark-sql-kafka-0-10_2.13:4.1.4. Feed the script on standard input as before. The first is the stateless query on 4.1.4, which printed HELLO for the first message and REAL and TIME in the same batch, and finished with exception=None:

import org.apache.spark.sql.streaming.Trigger
import org.apache.spark.sql.functions._
val boot = "localhost:9092"
val topic = "rtm_tarball_41"
def produce(msgs: Seq[String]): Unit = msgs.toDF("value").write.format("kafka").option("kafka.bootstrap.servers", boot).option("topic", topic).save()
produce(Seq("seed"))
val out = spark.readStream.format("kafka").option("kafka.bootstrap.servers", boot).option("subscribe", topic).option("startingOffsets", "latest").load().select(upper(col("value").cast("string")).alias("v"))
val q = out.writeStream.format("console").outputMode("update").trigger(Trigger.RealTime("5 seconds")).start()
println("RTM41 started active=" + q.isActive)
Thread.sleep(12000)
produce(Seq("hello", "real", "time"))
Thread.sleep(14000)
println("RTM41 exception=" + q.exception)
q.stop()
sys.exit(0)

The second is the stateful count on 4.3.0. It needs spark.sql.shuffle.partitions small enough that every stage can run at once, and local[6]. It printed the counts a 1, 2 and 3 and b 1 in one batch, and finished with exception=None:

import org.apache.spark.sql.streaming.Trigger
import org.apache.spark.sql.functions._
val boot = "localhost:9092"
val topic = "rtm_tarball_43"
def produce(msgs: Seq[String]): Unit = msgs.toDF("value").write.format("kafka").option("kafka.bootstrap.servers", boot).option("topic", topic).save()
produce(Seq("seed"))
val counts = spark.readStream.format("kafka").option("kafka.bootstrap.servers", boot).option("subscribe", topic).option("startingOffsets", "latest").load().select(col("value").cast("string").alias("word")).groupBy("word").count()
val q = counts.writeStream.format("console").outputMode("update").option("checkpointLocation", "/tmp/rtm_tarball_43_ckpt").trigger(Trigger.RealTime("5 seconds")).start()
println("RTM43 started active=" + q.isActive)
Thread.sleep(14000)
produce(Seq("a", "b", "a", "a"))
Thread.sleep(16000)
println("RTM43 exception=" + q.exception)
q.stop()
sys.exit(0)

Cleaning up

Delete the extracted folders when you are done. Each unpacks to about 600 MB, so four take roughly 2.4 GB, in addition to the tarballs.

Real-world scenarios and production failures

Almost every failure in this line falls into one of five shapes. Each one below was hit during the runs, with the exact text it produced, so you can grep your own logs for it.

1. The upgrade that turns a green job red

ANSI mode is on by default from 4.0, so work that silently produced NULL now stops:

[CAST_INVALID_INPUT] The value 'abc' of the type "STRING" cannot be cast to "INT"
because it is malformed. ... Use `try_cast` to tolerate malformed input and return NULL instead.

[DIVIDE_BY_ZERO] Division by zero. Use `try_divide` to tolerate divisor being 0 and
return NULL instead. If necessary set "spark.sql.ansi.enabled" to "false" to bypass this error.

The message names the fix. Use try_cast and try_divide where a NULL was the intended answer, and keep ANSI on elsewhere, because the bad rows it now rejects were wrong all along. Setting spark.sql.ansi.enabled=false is the escape hatch the error itself offers, and it restores the old behaviour for the whole session.

2. A feature that parses and then does nothing

The engine defines the syntax, and a connector has to implement it. The built-in file formats do not, so the statement is accepted and the run fails when it starts:

[AUTOCDC_TARGET_DOES_NOT_SUPPORT_MERGE] Cannot start AutoCDC flow: the target table
`spark_catalog`.`default`.`customers_history` (format: parquet) does not support
row-level operations.

That is the Auto CDC case on 4.3.0, reproduced with STORED AS SCD TYPE 2. The same shape applies to the CHANGES clause, INSERT ... WITH SCHEMA EVOLUTION, Data Source V2 transactions and @-syntax time travel in the “Also in” tables above. Before you plan around one of these, check that the connector you use implements it. The release note tells you the engine can; only the connector’s documentation tells you that your table can.

3. A feature that is off until you switch it on

Each of these failed in a run with a message that names what to set:

The last row is a different kind of failure: it is not disabled, it does not exist yet. A query using v:field on a 4.0 cluster fails to parse, so a job developed on 4.1 and deployed to an older cluster breaks at the first statement.

4. A Python environment that is not ready

PySpark features in this line depend on packages and interpreter versions that the official image and a stock machine do not provide.

5. Real-time mode that will not start, or starts and finds no slots

Stateful real-time mode runs every stage of the query at once, so the cluster needs a free task slot for each. With Spark’s default 200 shuffle partitions on a small session the first attempt failed:

[CONCURRENT_SCHEDULER_INSUFFICIENT_SLOT] Cannot run the pipelined stage group:
it needs 201 concurrent task slots to run all its stages together, but only 2
are currently free.

The cure is to size spark.sql.shuffle.partitions to what the cluster can run concurrently; 2, on a session with six cores, worked. Two further limits come from the sources rather than from a run. The source allowlist (RealTimeModeAllowlist, with spark.sql.streaming.realTimeMode.allowlistCheck defaulting to true) accepts the Kafka source, and console, memory, foreach and Kafka sinks. A query outside it fails with the error class STREAMING_REAL_TIME_MODE.OPERATOR_OR_SINK_NOT_IN_ALLOWLIST. And the 4.3 documentation states that ASOF JOIN is not supported in real-time mode.

Environment traps that look like Spark bugs

Four problems from the setup are worth repeating because each one produced plausible-looking output.

• A different Spark answered. A SPARK_HOME in the shell made the unpacked 4.0.5 tarball’s spark-shell report SPARK 2.4.4. Print spark.version before you trust a result.
• spark-shell -i did nothing. The REPL started and quit without error.
• Sketch results moved. kll_sketch_get_quantile_bigint returned 499 and 989 on 4.1.3 and 500 and 990 on 4.1.4 for the same input. A sketch is approximate, so assert a tolerance, not an exact value, in a test.
• A harness bug looked like a feature failure. A runner that skipped SQL after a heading produced “view not found” errors on statements that were fine. When a new feature “fails” on the first run, run the previous statement alone.

Optimisation, tuning knobs and best practices

The settings that gate or shape these features

These are the settings that appeared in the runs, with the default each release reported.

What each feature costs

Which upgrades are worth it on their own

This part is my judgement, drawn from what ran, not a measurement.

• To 4.0. Not optional once you leave 3.x, because Scala 2.13 and Java 17 are the baseline. The work is dependency and ANSI-mode testing, and the language gains (pipe syntax, SQL functions, session variables, collations) come with it.
• 4.0 to 4.1. Worth it for SQL scripting, recursive CTEs and VARIANT colon paths if you have logic living in driver code. Declarative Pipelines is worth a trial on one pipeline, with dry-run, before you commit a team to it.
• 4.1 to 4.2. QUALIFY and metric views pay back quickly for analysts. The geospatial types are a foundation: five functions today, no distance or containment.
• 4.2 to 4.3. JSON_TABLE, UNNEST and the VARIANT editing functions help any pipeline that handles nested documents. Stateful real-time mode is for the few queries that need sub-second latency and a Kafka source. 4.3.0 shipped without preview builds, so a cautious team waits for the first patch release.

Best practices

1. Test the upgrade in this order: ANSI errors on your real data first, then the settings table above for anything a job relies on, then the Python environment.
2. Pin the interpreter and the packages with the release: the Python version floor moved from 3.9 to 3.11 across the four releases.
3. Prefer one flag at a time. A feature behind a flag should be switched on in the session that uses it, not cluster-wide, so a failure points at the feature.
4. Assert tolerances for sketches, and exact values for everything else.
5. Check the connector before the engine. For Auto CDC, CHANGES, schema evolution and transactions, the connector decides whether the feature exists for your table.
6. Read spark.version at the top of every run. It is the cheapest check against running the wrong install.

Summary and key takeaways

The mental model is three decisions across four releases. SQL became the primary surface again, with pipe syntax, scripting, recursive CTEs, QUALIFY, NEAREST BY, ASOF JOIN, JSON_TABLE and UNNEST. Batch and streaming moved toward one declarative layer, with pipelines, change data capture and a real-time mode that gained state in the newest release. Python and Spark Connect moved off the JVM boundary, with Arrow by default and a thin client.

• Know the maturity of what you adopt. A feature is Default, GA, behind a flag, preview, or dependent on a connector, and those are not interchangeable.
• The cheap failures name their fix. ANSI errors, disabled features and missing flags all say what to set. The expensive ones are quiet: a connector that does not implement a feature, a wrong Spark on the path, a sketch that moved.
• Many headline features need something outside Spark. pyarrow, pyyaml, Kafka, a merge-capable connector, a newer Python.
• The newest release has had the least outside testing. 4.3.0 shipped without previews, so give it the same scrutiny a preview would get.
• Run the example before you rely on it. Every claim in this post comes from a run, and four of the problems the runs found were in the harness, not Spark.

The line is moving faster now than at any point in the 4.x series: the last gap between releases was about 3.6 months, half the earlier two. Treat the feature tables here as a snapshot, re-check the maturity tags against the release note of the version you deploy, and keep the SQL files from the hands-on section as a smoke test you can rerun on the next one.

References

• Spark 4.0.0 release notes, 4.1.0, 4.2.0 and 4.3.0 for the feature lists and ticket numbers used throughout
• Every Apache Spark release, and the problem each one was built to solve for removals, changed defaults and upgrade risks
• Spark Connect: the driver becomes a server for the client and server split behind the thin client and the pipelines CLI
• Structured Streaming internals for state and checkpoints behind real-time mode
• JOIN syntax, including NEAREST BY, QUALIFY and SET PATH in the 4.2.1 documentation
• JSON_TABLE, ASOF JOIN and UNNEST in the 4.3.0 documentation
• Spark Declarative Pipelines programming guide for the project spec, the CLI, and Auto CDC with its target requirements
• RealTimeModeAllowlist.scala for the sources, sinks and operators real-time mode accepts
• Apache Spark downloads and archive for the release tarballs and their checksums

Trademarks

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