DEV Community: lnc

Apache Iceberg v4 Column Updates Are About to Change Feature Engineering

lnc — Mon, 09 Nov 2026 20:03:03 +0000

Efficient Column Updates in a Nutshell

Apache Iceberg has two common ways to change a row in a table, and both of them work at file or row granularity. Under copy-on-write, the engine finds the data file that holds the row, rewrites the entire file with the new value, and swaps the old file out of the table's metadata. Under merge-on-read, it leaves the old file alone, records the old row as deleted, and writes the new row into a new data file; readers apply the deletes while scanning. In the v2 version of my experiment, that delete was a position delete file (a list of file paths and row numbers). In the v3 version, it was a deletion vector, or DV (a compressed bitmap of positions for one data file). Either way, Iceberg is still changing rows, not individual columns.

Row-level updates are good enough for many workloads. They get expensive on very wide tables: feature stores, embedding tables, and model-scoring tables with hundreds or thousands of columns. A nightly job might recompute one feature, refresh one model's scores, or backfill one new column for every existing row. That is one column changing across the table, but today it can still mean rewriting every byte in the touched data files.

Efficient column updates in Iceberg is the v4 work aimed at that case. It builds on an earlier column-family proposal, where related columns would have been declared up front and written separately from the start. Column updates keep that idea in view, but the focus is narrower: make the cost of a write closer to the columns that changed instead of the current all-or-nothing shape.

At the heart of the design is a new column file: a file that holds one or more whole columns for every row of an existing data file. That lets the data payload for a column update be the updated column, not the whole row. Iceberg finds column files the same way it finds everything else, through its manifests: metadata files that list every data file in the table along with row counts, partition values and column statistics. A column file is recorded on the manifest entry for the data file it belongs to. The reader then uses that metadata to turn the base file plus its column files back into rows.

Current Column Updates in Iceberg Are Expensive

Take a table shaped like a feature store: an id column plus two hundred double columns of random numbers, with 200,000 rows. I wrote it with Spark 4.0.1 and Iceberg 1.12.0 as a single 300.9 MB Parquet file, so every update has one base file to deal with and the snapshot history stays easy to read. Then I made two copies: one with write.update.mode set to copy-on-write, and one set to merge-on-read. I also ran the same experiment on v2 tables. The only difference was that v2 wrote a position delete file where v3 wrote a deletion vector, so I'm only showing v3 here.

Against each copy, I ran two UPDATE statements. The first is the backfill case: change one column, c_1, for every row. The second is the partial case: change one column, c_2, for one row in ten. After each statement, the newest snapshot summary and the .files metadata table show what the commit added and how large it was.

-- the backfill case: every row, one column of two hundred
UPDATE local.db.wide_v3_merge_on_read SET c_1 = c_1 + 1;

-- the partial case: one row in ten, one column
UPDATE local.db.wide_v3_merge_on_read SET c_2 = c_2 + 1 WHERE id % 10 = 0;

-- what I read after each one
SELECT summary FROM local.db.wide_v3_merge_on_read.snapshots ORDER BY committed_at DESC LIMIT 1;
SELECT content, record_count, file_size_in_bytes FROM local.db.wide_v3_merge_on_read.files;

Update mode	Statement	What the commit added	MB written
copy-on-write	all rows, `c_1`	one new data file, all 200 columns (the old file deleted)	301.1
merge-on-read	all rows, `c_1`	one new data file, all 200 columns, plus one deletion vector	301.1 + under 0.01 (DV)
copy-on-write	one row in ten, `c_2`	one new data file, all 200 columns (the old file deleted)	301.1
merge-on-read	one row in ten, `c_2`	one new data file of 20,000 rows, all 200 columns, plus one deletion vector	30.1 + 0.03 (DV)
column file (v4, estimated)	all rows, `c_1`	one column file holding `c_1` for every row, plus a rewritten manifest entry	about 1.6
column file (v4, estimated)	one row in ten, `c_2`	one column file holding `c_2` for every row (nine tenths carried over), plus a rewritten manifest entry	about 1.6

The bold rows are estimates. No released Iceberg can run a v4 column update.

For the backfill, both modes wrote a new 301.1 MB file, slightly larger than the 300.9 MB base file because the updated values compress a little differently. That is all 200 columns rewritten to change one. Copy-on-write does this by definition. Merge-on-read does it because Iceberg treats an UPDATE as a delete of the old row plus an insert of the new one. When every row changes, the insert half is a complete new data file. The deletion vector marks every row of the old file as gone, and until compaction the table holds both files.

The partial update is where the two modes differ. Merge-on-read wrote a 30.1 MB file holding every column of the 20,000 touched rows, about a tenth of the table. Copy-on-write rewrote the full 301.1 MB file again. Merge-on-read scales with the number of rows touched, but it still writes every column for each touched row.

A v4 column update changes the scale of the write. The column c_1 on its own, with the same values in the same order, is 1.6 MB as a Parquet file: about half a percent of the base file.

In the current modes, write size is driven by rows touched times columns carried. Changing one column or 200 columns has the same shape.

Byte by Byte

For the data payload, a v4 column update writes one new file holding the updated column for every row. The dominant write is that one column. On this table every column is random doubles, so each column is roughly 1/200th of the file:


base file (200 columns, 200,000 rows)	300.9 MB
one column, by proportion (300.9 MB ÷ 200)	about 1.5 MB
`c_1` written on its own as a Parquet file, measured	1.6 MB

The partial update lands at the same estimate. A column file must hold every row of the column, so changing one row in ten still writes the column in full. The untouched nine tenths are copied from the base file. The size depends on the column, not on how many rows changed.

Two small costs are left out of that estimate: the rewritten manifest entry, which is a fraction of a megabyte here, and a null at each deleted position if the base file has deletes. At typical deletion ratios, neither changes the rough size of the write.

The expected write drops from 301.1 MB to about 1.6 MB: nearly 200x on this table because the table has 200 similarly sized columns. For any table, the factor is the base file size divided by the updated column's size. column_sizes in the .files metadata table gives that size per column. A real writer will add overhead on top of the column file; this demo does not measure that yet.

Anatomy of a Column File

The public v4 draft defines a column file as "a file that stores the values of one or more fields for each row of another file." That last part matters. A column file is not a list of changed cells. It has the same row count as the data file and holds the complete current value of one or more fields for every row. The data file's manifest entry lists which fields live in which column file:

{
  "location": "s3://bucket/data/base_file_1.parquet",
  "record_count": 1000,
  "column_files": [
    {
      "location": "s3://bucket/data/column_update_7.parquet",
      "field_ids": [7],
      "file_format": "parquet",
      "file_size_in_bytes": 12345
    }
  ]
}

The sketch omits most tracked-file fields. The important part is the column_files list. Each entry names the column file's location, the live top-level field IDs it stores, the file format (avro, orc or parquet) and the file size in bytes. It may also carry encryption key metadata.

A reader that understands column files resolves each projected field from the column file that lists it and every other field from the data file. The remaining requirements:

Requirement	Consequence
It is stored in a normal data-file format	The bytes are ordinary Parquet, ORC or Avro; stitching them into the base file still requires Iceberg v4 column-file support.
Whole top-level fields only	A struct is updated as a unit; one leaf cannot be updated alone.
One column file per field	Updating a field again replaces its column file instead of stacking another.
Every row, in the base file's order	The read is a column swap by position; the write has to line up exactly.
Unchanged rows are carried over	A partial update still writes the whole column.
Deleted rows are written as null	A heavily deleted file carries a few percent of filler.
The deletion vector decides what is deleted	A null is a filler, never a tombstone.

The fifth requirement sets the size of a partial update. A column file for one row in ten is not 20,000 values keyed by position. It is 200,000 values: 20,000 new values and the rest copied from wherever the field lived before, or null where the row is deleted.

That is the 1.6 MB file from the demo, not a 0.03 MB delta.

A column file is the whole column, not a delta. It mirrors its base file row for row and is swapped in by position at read time. Wide backfills get cheaper; row-level refreshes are not the sweet spot.

Note

"Column families" is a different thing
The earlier proposal was about column families: declaring which columns belong together so they are written as separate files from the start. The design doc keeps that as future work and the current column-file text does not define families. "Column file" in this post is the v4 spec term.

Stitching It Back Together on Read

A read of a data file with column files has four steps:

Use the manifest entry to split the requested fields between the base file and column files.
Open the base file and only the column files needed for the projection.
Read those files in position order and combine each position's pieces into one row or batch.
Apply the deletion vector to the combined result.

Because the files are row-aligned, step 3 is what the design doc calls "reference substitution": the column file's array replaces the base file's array for the updated field. The intended read shape is a column-level substitution, not a join or keyed merge.

In the Java implementation work, step 3 is a stitcher. There is one stitcher per engine row type, such as Spark InternalRow, Flink RowData, and the generic Record. Steps 1, 2 and 4 are shared across engines.

A column file is opened only when a query projects one of its fields. A query on c_200 reads the base file alone.
Column files do not need to share the base file's row groups, the chunks of rows a Parquet file is written and read in. When row groups do not line up, the reader slices batches to the overlapping window.
Deletion vectors are applied after stitching, so a column file can hold nulls at deleted rows and is never consulted about deletion.

Explaining every row

The every-row requirement is what makes partial updates less dramatic.

The alternative was a sparse column file: only the changed rows, each tagged with its position. A one-in-ten update would then weigh a tenth of the column. The design rejects that tradeoff because it moves cost to readers outside the JVM:

The two representations have asymmetric read-side cost across language ecosystems. Spark's ColumnVector is an abstract class with virtual dispatch, so a custom stitching subclass can redirect per-row reads (getInt(rowId) → choose base vs column file) with no memory copy. Concrete Arrow array types in arrow-rs, PyArrow, and arrow-go do not support this yet.

In Spark, a sparse file can stay lazy on read. A ColumnVector subclass can answer getDouble(i) from the column file when position i was updated and from the base file otherwise, without materializing a replacement array.

An Arrow array in Rust, Python or Go is a contiguous buffer with no per-row redirect. Those readers would have to materialize a new array per batch by scattering the updated values over the base column. That read cost gives back much of what the sparse file saved on write.

With every row present, the stitch can be expressed as an array-level substitution instead of a sparse overlay.

v4 tables need a layout that works across those engines, so the draft standardizes the every-row representation instead of letting writers choose between dense and sparse column files. Once a layout is allowed, every reader has to handle it.

	Every row (the spec)	Updated rows only (the alternative)
Partial update writes	the whole column	only the changed rows
Read in Spark	swap one array	cheap either way
Read in Rust, Python, Go	substitute an array	copy the column

How a column update effects manifest

A column update changes the manifest as well as the data. Conceptually, the new snapshot stops pointing at the old version of the tracked file and points at a successor entry for the same base file with a new column_files list. Earlier snapshots still see the old metadata for time travel. A field may appear in only one column file for a tracked file, so updating one field of an existing column file means rewriting that column file with its other fields carried over.

The current tracked entry also gets a new sequence number: the file's age in commit order, which decides which delete files apply to it. With the new number, older deletes should not be replayed against the new column values.

That sequence-number change would otherwise make older deletes tricky. The v4 work handles this by folding deletes that already applied to the file, including equality deletes, into the entry's own deletion vector. The DV becomes the complete delete record for the current version of the tracked file, and the reader does not need to replay old delete files against the updated columns.

Stats for the updated fields are recomputed from the column file, so file pruning based on those fields can still work.

Row lineage is part of this too. Since v3, every row carries _last_updated_sequence_number, the sequence number of the commit that last changed it. One of the new column files has to hold that field: null for every row the update changed, so those rows inherit the new sequence number on read, and the previous value for every row it did not change. Row IDs do not change.

If the updated field is used for partitioning or sorting, metadata has to become conservative for that file because the rows may no longer match the old placement information. Plan for weaker pruning and ordering metadata on that file until compaction rewrites it.

Step	Cost
Read the column's current values and the file's deletes	a read before the write, even for a partial update
Write one column file per base file	the column file
Rewrite the manifest entry, stats and partition info	small, but every touched file gets it
Compact later (`rewrite_data_files` folds column files back in)	stitching is a cost between compactions, not forever

But why not directly in Parquet?

The design doc evaluated a Parquet-level alternative and discarded it. Parquet's footer has a ColumnChunk.file_path field, so in theory a Parquet file could say that its score chunk lives in one file while its other chunks live in users.parquet. A column update could then be a narrow Parquet file whose footer describes the whole row. Iceberg would record it as an ordinary data file, with no spec change.

The catch is in the comment on that field in today's parquet.thrift: "there are no known reference implementations that will read externally stored column data if this field is populated within a standard parquet file. Making use of the field for this purpose is not considered part of the Parquet specification." The field exists, but readers do not honor it. "No Parquet change" would have become a parquet-format change after all, followed by changes in every Parquet reader Iceberg depends on.

	In the manifest (chosen)	In the Parquet footer (discarded)
Who must change	Iceberg's spec and readers	parquet-format and every Parquet reader
Can the planner skip files?	yes, from the manifest	only after opening footers
Can cleanup jobs see every file?	yes	no, the link is inside a footer
Formats	Parquet, ORC, Avro	Parquet only

The File Format API makes the same kind of choice: keep the bytes in plain data-file formats and put the table-level structure in Iceberg metadata, where every engine already has to look.

A column file is tracked by Iceberg, not by Parquet. v4-aware expire_snapshots, remove_orphan_files and rewrite_data_files implementations can account for column files from the manifest. A Parquet tool that opens a column file directly sees an ordinary narrow Parquet file with no link to its base file.

State of the Implementation

Everything above is based on the public v4 draft, the proposal issue and open implementation work as of 2026-10-27. The public spec includes v4 column-file text, but v4 itself is still under active development and has not been formally adopted. The row-count requirement has changed once already; implementation details can still move.

Piece	State	Where
Proposal	open, milestone "Iceberg V4 Spec"	issue 15146
Design doc	current	doc
Spec text	present in the public v4 draft, still part of active v4 work	Iceberg spec and linked PRs
Readers and writers	implementation work in progress	linked under Further reading
Format version 4	not formally adopted; nothing here is in Iceberg 1.12.0

The demo ran once on one machine, measured file sizes only, and used random doubles that make every column about the same width. The reference column is a Spark Parquet file, not an Iceberg column file, so a real column file may be slightly larger. The filler-overhead figures come from the design doc and were not reproduced here. I did not build the reader PR. The research, demo script and draft used AI assistance; the reading of the spec and the opinions are mine, and so are any mistakes.

What you should know and how to proceed

You have a wide table with a regular column refresh. Budget for the full rewrite until v4 ships in your engine, or keep the fast-moving columns in a narrower table on the same key and join at read time. That is column families by hand. Migrating to v4 later becomes a consolidation, not a full redesign.
Your refresh touches a few rows, not all of them. Stay on merge-on-read unless the math says otherwise. A column file writes every row of the column, so compare the fraction of rows you touch against the column's share of bytes (column_sizes in the .files table) before expecting v4 to help.
You run compaction and orphan cleanup on a schedule. Every query on an updated column reads the base file and the column file until rewrite_data_files folds them back together. Compaction is part of the column-update workflow, not an optional cleanup later.
You partition or sort on a column you might update. Treat this as a conservative-metadata case. Pruning and ordering information can get weaker for the affected file until compaction rewrites it.
You write an engine. The stitcher is the interface to watch. Every-row column files make the stitch an array-level substitution; Arrow arrays are the reason the layout is every-row.

Spark Connect from Rust: Pull Spark outputs into DuckDB and Polars

lnc — Mon, 09 Nov 2026 20:02:57 +0000

Setting the Rust-Native Scene

In the OCaml post we walked through the process of building a Connect client from scratch, but in truth the Rust client is probably the only one you really need if you're working right now in the data engineering field. It gives you in-process interoperability with engines like Polars (written in Rust) and DuckDB (C++, with official Rust bindings), meaning you no longer have to choose between Spark and . In my connect idiot's guide I only hinted at the Rust client as part of a way to gauge supported operations. In this follow-up, I want to deep dive into the Rust Client. It'll be helpful to know what Spark connect is ahead of time, but you'll probably figure it out pretty easily if you're a smart cookie. TLDR on the previous posts: Connect is Spark's attempt to decouple the JVM from the user's environment by implementing a client-server architecture where a very lightweight client sends unresolved logical plans to a remote cluster.

Spark Connect sends query results to the client as Arrow record batches (thank god). The client builds a plan, the server runs it, and results are returned as Arrow. If you want to know why this is important, I'd suggest reading my previous blog on Arrow batches replacing pickle serialization between the JVM and Python process in Spark. Every Connect client gets those same Arrow batches, so this isn't unique to Rust: in Python, PySpark's toArrow() (4.0 and later) hands you a pyarrow table that DuckDB and Polars can read directly. What Rust adds is doing it in one compiled program with no Python in the process, where DuckDB and Polars read the batches as Rust libraries. This separation allows Spark to do the work that needs a cluster (the scan, the join across a billion rows, the group-by), and when the result comes down as Arrow, DuckDB or Polars can be used in the same process without writing intermediary data files.

The demo and code in this blog can be found in spark-demos/demos/09_spark_connect_rust with its full run log.

With the official Rust client, one Spark result comes out as Arrow IPC bytes that DuckDB and Polars both read cheaply, DuckDB registers it without a further copy, and the answer goes back to Spark.

What the Rust client actually is

apache/spark-connect-rust is the Apache Spark project's own Rust client, listed on the Spark Connect page next to the Go and Swift ones, with its documentation at apache.github.io/spark-connect-rust. The crate is apache-spark-connect, used in code as spark_connect, and the version number is the Spark version it speaks: 4.2.0 means Spark 4.2.0 and later, which is a nice convention more clients should steal.

The architecture page describes one Rust core (plan building, gRPC via tonic, Arrow decoding) with two faces on it: a synchronous Rust API shaped like PySpark, and a pyspark-client-rust wheel on PyPI that installs as pyspark and is meant as a drop-in for the reference Python client. This blog uses the Rust API. If you've used other Rust gRPC clients, "synchronous" is the part to notice: no async, no runtime to set up, and no .await anywhere in your code. The hello world from the docs is the entire API in miniature:

use spark_connect::{SparkSession, functions as f, lit};

fn main() -> Result<(), Box<dyn std::error::Error>> {
    let spark = SparkSession::builder()
        .remote("sc://localhost:15002")
        .get_or_create()?;

    let df = spark
        .range(1_000_000)?
        .select([(f::col("id") * lit(2)).alias("x")])
        .filter((f::col("x") % lit(3)).eq(lit(0)));

    println!("count = {}", df.count()?);
    Ok(())
}

The compatibility page says the goal is full API parity with the reference PySpark Connect client, that the protobuf plans it builds are checked byte-for-byte against the reference Python client in CI, and that the official PySpark Connect test suite is run against it. In practice, if you know PySpark you can mostly guess the Rust method name (spark.sql(), df.filter(), df.group_by(), df.write().save_as_table()), and when you guess wrong the compiler tells you, which is more than Python does.

Setting up the environment

The server. Any Spark 4.2 or later Connect server will do. The demo's compose.yaml starts the official image's Connect server (it listens on all interfaces by default) and gives it a health check, so docker compose up --wait returns when the port actually answers rather than when the container starts:

services:
  spark-connect:
    image: apache/spark:4.2.0-scala2.13-java21-python3-ubuntu
    entrypoint:
      - /opt/spark/sbin/start-connect-server.sh
      - --wait
      - --conf
      - spark.driver.memory=4g
    ports:
      - "127.0.0.1:${CONNECT_PORT:-15002}:15002"
    healthcheck:
      test: ["CMD", "bash", "-c", "</dev/tcp/127.0.0.1/15002"]
      interval: 5s
      retries: 40

docker compose up -d --wait

The published port is the one thing you may need to change. My machine already had a Spark Connect server on 15002 from another demo the first time I ran this (this keeps happening to me; it's the first item in the idiot's guide's debugging section), so that run used CONNECT_PORT=15902. It's also why the first thing the program prints is the server's version.

The crate. Three things have to be installed before cargo build works, and each fails at build time if you skip it. You need protoc, because the client compiles the Spark Connect .proto files when it builds; a C++ compiler plus CMake, because DuckDB is built from source; and Rust 1.98, because polars 0.55 uses standard library APIs stabilised after 1.92 (std::hint::cold_path among them) and won't compile on anything older. The demo pins 1.98 in rust-toolchain.toml, so rustup fetches the right compiler on its own, and the other two are one package each:

sudo apt-get install protobuf-compiler build-essential cmake   # Debian/Ubuntu
brew install protobuf cmake                                    # macOS

Then this Cargo.toml, which is the part worth copying exactly. The polars feature list is the demo's own needs (lazy frames, IPC, rolling windows) plus four I found the slow way on 0.54 so you don't have to: parquet, strings and temporal, without which lazy doesn't compile, and timezones, without which a timestamp column from Spark won't print:

[dependencies]
# The official client. Its `polars` feature (to_polars()) is left off: it pins polars 0.54.
spark-connect = { package = "apache-spark-connect", version = "4.2" }

# arrow-rs (the Rust Arrow library), at DuckDB 2.0's major. The client uses 58 internally.
arrow = { version = "60", features = ["prettyprint"] }

# duckdb-rs for DuckDB 2.0, statically linked. `vtab-arrow` lets DuckDB query arrow-rs batches.
duckdb = { version = "1.20000.1", features = ["bundled", "vtab-arrow", "appender-arrow"] }

# The newest polars crate. It reads Arrow IPC with its own Arrow implementation.
polars = { version = "0.55.2", features = [
    "lazy", "ipc", "ipc_streaming", "dtype-full", "fmt", "timezones",
    "rolling_window", "pct_change", "round_series", "parquet", "strings", "temporal",
] }

Two version facts are worth knowing before you touch that file. DuckDB 2.0 came out on 2026-10-21, and its Rust crate (duckdb 1.20000.1; the crate encodes DuckDB 2.0.0 in the minor version) is built on arrow-rs 60, while the client is still on 58. Different arrow-rs majors are different Rust types, so a RecordBatch from the client won't type-check where DuckDB wants one; the hand-off is the Arrow IPC bytes from to_arrow(), which any arrow-rs major can read. Polars doesn't use arrow-rs at all; it ships its own Arrow implementation (polars-arrow) and reads the same IPC bytes. The client's to_polars() does exactly that internally, but against polars 0.54, so the demo leaves that feature off and calls Polars' IPC reader itself, which works with whatever polars you have. As for Polars 2.0, it's a Python release (2.0.0 on 2026-10-06, 2.1 since). There's no 2.x polars crate yet, and the 2.x source only builds with Polars' own patched hashbrown and a nightly toolchain, so the newest crate, 0.55.2, is what a Rust program gets. (The client's own installation page says Rust 1.70 or later, which is true of the client alone; Polars is what raises the floor.) The first build took 4m23s in a clean container, since DuckDB compiles from source, and a rebuild after a code change takes seconds.

Step 1: Some data to play with

Nothing is downloaded. Spark builds a 20-million-row orders table from range() and hash functions, so the whole demo runs anywhere a Spark 4.2 server is reachable and nobody has to find a dataset:

let spark = SparkSession::builder().remote("sc://localhost:15002").get_or_create()?;
println!("Spark {}", spark.version()?);

spark.sql(
    "CREATE TABLE IF NOT EXISTS orders USING parquet AS
     SELECT id AS order_id,
            element_at(array('sfo','lax','sea','nyc','chi','aus','den','pdx','phx','mia','bos','atl'),
                       CAST(pmod(xxhash64(id, 7), 12) AS INT) + 1)            AS city,
            timestamp_seconds(1756684800 + pmod(xxhash64(id, 1), 2592000))     AS ts,
            CAST(round(5 + pmod(xxhash64(id, 2), 9500) / 100.0, 2) AS DOUBLE)  AS amount,
            CAST(1 + pmod(xxhash64(id, 3), 6) AS INT)                         AS items,
            element_at(array('delivered','delivered','delivered','cancelled','refunded'),
                       CAST(pmod(xxhash64(id, 4), 5) AS INT) + 1)             AS status
     FROM range(20000000)",
)?
.collect()?;

The .collect() on the end is not a typo. Everything in this client is lazy, the same as PySpark: spark.sql() builds a plan, and nothing runs on the server until an action asks for a result, which for a CREATE TABLE means you have to ask for its (empty) result on purpose.

Spark 4.2.0 at sc://localhost:15002
  orders: 20000000 rows

Step 2: Getting a result out as Arrow (four ways, and one of them is wrong)

The client has four ways worth knowing to turn a DataFrame into something you hold in Rust (there are others, such as to_local_iterator() for streaming rows and to_datafusion() behind a feature flag), and picking the right one is most of the performance story in this blog:

Method	You get	Use it when
`collect()`	`Vec<Row>`, each value boxed as a `Value` enum	You want to look at a few rows
`collect_record_batches()`	`Vec<arrow::RecordBatch>`, the server's batches as decoded	The result is going to DataFusion or your own arrow-rs 58 code
`to_arrow()`	`Vec<u8>`, one Arrow IPC file	The result is going to a file, a socket, or another Arrow library
`to_polars()`	`polars::DataFrame` (polars 0.54)	The result is going to Polars and you're on 0.54; otherwise read `to_arrow()` with Polars' `IpcReader`

One city's orders, as batches and as one IPC file (the form DuckDB and Polars get below):

let sfo = spark.sql("SELECT order_id, city, ts, amount, items, status FROM orders WHERE city = 'sfo'")?;
let batches = sfo.collect_record_batches()?;
let ipc = sfo.to_arrow()?;

176 RecordBatches, 1667285 rows
arrow-rs schema: order_id: Int64, city: Utf8, ts: Timestamp(µs, "Etc/UTC"),
                 amount: Float64, items: Int32, status: Utf8
as Arrow IPC: 77.3 MiB

The batch count is the server's choice, not yours. Connect caps each batch at spark.sql.execution.arrow.maxRecordsPerBatch rows (default 10,000) and at a byte size, spark.connect.grpc.arrow.maxBatchSize (default 128 MiB in Spark 4.x, of which it uses 70%), and the client hands you the batches exactly as they arrived. Here the row cap is the one that bites (176 batches rather than the 167 the cap alone would give, since a batch never spans two partitions), and arrow::compute::concat_batches joins the pieces into one when a consumer wants that.

The same 1.67 million rows, each way, on one machine with the server in a container on the same host. Each path re-runs the query on the server, and each number includes that scan (an earlier count and sum on the same table had already put the files in the OS page cache and warmed up the JVM):

Path	ms
`collect_record_batches()`	683
`to_arrow()`	477
`collect()`	1475

Rows cost about twice what batches cost for the same data. Most of the difference is client-side work. (These are single runs: to_arrow() calls collect_record_batches() internally and still came in faster, so the noise is a couple of hundred milliseconds; the doubling for rows is well outside it.) In the client's source, collect() decodes the same Arrow batches and then builds a Row for every row and a Value for every cell, which is a perfectly reasonable thing to do for twenty rows you intend to print and a silly thing to do for 1.67 million you intend to hand to DuckDB. If the result is going to another engine, it should never pass through collect().

Step 3: DuckDB 2.0 gets the batch

First the result crosses from the client's arrow-rs to DuckDB's: the to_arrow() bytes from step 2, read with arrow 60's IPC reader and joined into one batch. For the 1.67 million rows that takes 70 ms:

use arrow::ipc::reader::FileReader;   // arrow 60, DuckDB's major

fn ipc_to_batch(ipc: &[u8]) -> Result<RecordBatch, Box<dyn Error>> {
    let reader = FileReader::try_new(Cursor::new(ipc), None)?;
    let schema = reader.schema();
    let batches: Vec<RecordBatch> = reader.collect::<Result<_, _>>()?;
    Ok(concat_batches(&schema, &batches)?)
}

duckdb-rs has an Arrow table function behind the vtab-arrow feature. In DuckDB 2.0's crate you register a RecordBatch by wrapping it in an ArrowBatchRegistration, which you pass to arrow(?) as a query parameter. The batch isn't copied; it stays registered until the registration is dropped. A view can't take a prepared parameter, so to get a view the registration goes into a DuckDB variable first:

use duckdb::vtab::arrow::{ArrowBatchRegistration, ArrowVTab};

fn arrow_view(db: &Connection, name: &str, rb: RecordBatch) -> duckdb::Result<ArrowBatchRegistration> {
    let reg = ArrowBatchRegistration::new(rb);
    db.execute(&format!("SET VARIABLE {name}_batch = ?"), [&reg])?;
    db.execute_batch(&format!(
        "CREATE VIEW {name} AS SELECT * FROM arrow(getvariable('{name}_batch'))"
    ))?;
    Ok(reg)
}

let db = Connection::open_in_memory()?;
db.register_table_function::<ArrowVTab>("arrow")?;
let _orders = arrow_view(&db, "orders", ipc_to_batch(&ipc)?)?;
let _surge = arrow_view(&db, "surge", surge_schedule(&ts_type)?)?;   // built in Rust; its timestamps use Spark's ts type

Keep the registrations in variables for as long as you query the views. Dropping one unregisters the batch, and the view then fails with is not registered rather than reading freed memory, which is the right failure but a confusing one if you discarded the return value with a ?;. (DuckDB 1.5's crate took two integers and a pair of ::UBIGINT casts instead, and kept every batch alive for the life of the process; the new API is an improvement on both counts.)

Now the reason I brought DuckDB in at all. The surge table is a per-six-hour price multiplier that exists only in the client process, and each order needs the latest multiplier at or before its timestamp. (It's built in Rust, with its starts_at column given the same Arrow timestamp type and zone as Spark's ts, so DuckDB sees two TIMESTAMP WITH TIME ZONE columns rather than one with a zone and one without.) That is an ASOF join (match each row to the nearest earlier key instead of an equal one), which DuckDB has as a first-class SQL operator. Spark has an as-of join inside it (pandas-on-Spark's merge_asof, and an AsOfJoin relation in the Connect protocol), but 4.2 has no SQL syntax for it (4.3 adds ASOF JOIN, off by default) and the Rust client doesn't expose it, so from Rust today this part is DuckDB's job:

WITH surged AS (
    SELECT o.ts, o.amount, s.multiplier, o.amount * s.multiplier AS surged
    FROM orders o
    ASOF JOIN surge s ON o.city = s.city AND o.ts >= s.starts_at
)
SELECT CAST(CAST(ts AS TIMESTAMP) AS DATE) AS day,
       count(*) AS orders, round(sum(amount), 2) AS revenue,
       round(sum(surged), 2) AS surged_revenue, round(avg(multiplier), 4) AS avg_multiplier
FROM surged GROUP BY 1 ORDER BY 1

+------------+--------+------------+----------------+----------------+
| day        | orders | revenue    | surged_revenue | avg_multiplier |
+------------+--------+------------+----------------+----------------+
| 2025-09-01 | 55829  | 2933784.48 | 3482949.81     | 1.1868         |
| 2025-09-02 | 55924  | 2927311.35 | 3476037.04     | 1.1881         |
| 2025-09-03 | 55491  | 2913342.93 | 3459069.6      | 1.1877         |
...
  PASS  DuckDB count == Spark count: 1667285
  PASS  DuckDB sum(amount) == Spark sum(amount): 87512653.61 vs 87512653.61
  PASS  ASOF JOIN kept every order: 1667285

The double cast on ts is the one seam in this step. Spark's timestamp arrives tagged with the session time zone (Timestamp(µs, "Etc/UTC")), DuckDB maps that to TIMESTAMP WITH TIME ZONE, and a direct cast to DATE fails with Conversion Error: Unimplemented type for cast (TIMESTAMP WITH TIME ZONE -> DATE). Zone-aware casts are what DuckDB's ICU extension provides, and the bundled build doesn't include it (duckdb-rs has an icu feature that does). Casting through plain TIMESTAMP sidesteps it for a UTC session; for any other zone, add the feature or do the date arithmetic in Spark before collecting.

One practical rule for this step, and it's the one that surprised me. Registering the batch is free, and a count and sum through the view take 65 ms. But the identical ASOF rollup took 1901 ms through the view and 139 ms on a native table that cost 146 ms to create from the same view:

CREATE TABLE orders_native AS SELECT * FROM orders;   -- 146 ms for 1.67M rows

So if you're going to run more than one query over the rows, or any join, copy them into a table first; the copy pays for itself on the first join. I had assumed the zero-copy view was the fast path because zero-copy sounds like the fast path, and it isn't. I haven't profiled why. Part of it is in the duckdb-rs source: every scan through the table function converts the Arrow vectors into DuckDB's own format again, so each query pays the copy that the native table paid once. My guess for the rest is that the view is a single sequential source with no statistics. The gap was the same on DuckDB 1.5.6 (2226 ms against 177), so 2.0 didn't change this rule.

Step 4: Polars gets Spark's aggregate

Spark is the right place for the 20-million-row group-by. Polars is the nicer place for what comes after, which here is a per-city rolling window and a day-over-day change, both of which are one expression in Polars and a window specification you'd have to think about in Spark SQL. to_arrow() runs the query and gives you the result as an in-memory Arrow IPC file, and Polars' IpcReader turns that into a polars::DataFrame. That's all to_polars() does inside the client; doing it yourself means it works with polars 0.55. It costs a copy into IPC bytes and a decode into Polars' own buffers, because Polars has its own Arrow implementation.

let daily = spark.sql(
    "SELECT city, CAST(ts AS DATE) AS day, count(*) AS orders, round(sum(amount), 2) AS revenue
     FROM orders GROUP BY 1, 2",
)?;
let pl_daily = IpcReader::new(Cursor::new(daily.to_arrow()?)).finish()?;   // Spark runs the group-by; 360 rows cross as IPC

let rolling = pl_daily.lazy()
    .sort(["city", "day"], SortMultipleOptions::default())
    .with_columns([
        col("revenue")
            .rolling_mean(RollingOptionsFixedWindow { window_size: 7, min_periods: 1, ..Default::default() })
            .over([col("city")])?
            .alias("revenue_7d"),
        col("revenue").pct_change(lit(1)).over([col("city")])?.alias("revenue_pct"),
    ])
    .collect()?;

polars schema: {"city": String, "day": Date, "orders": Int64, "revenue": Float64}
┌──────┬────────────┬────────┬───────────┬────────────┬─────────────┐
│ city ┆ day        ┆ orders ┆ revenue   ┆ revenue_7d ┆ revenue_pct │
│ str  ┆ date       ┆ i64    ┆ f64       ┆ f64        ┆ f64         │
╞══════╪════════════╪════════╪═══════════╪════════════╪═════════════╡
│ atl  ┆ 2025-09-01 ┆ 55524  ┆ 2.9116e6  ┆ 2.9116e6   ┆ null        │
│ atl  ┆ 2025-09-02 ┆ 55327  ┆ 2.9054e6  ┆ 2.9085e6   ┆ -0.002129   │
│ atl  ┆ 2025-09-03 ┆ 55638  ┆ 2.9335e6  ┆ 2.9168e6   ┆ 0.009665    │
...
  PASS  Polars sum(orders) == Spark count(orders): 20000000

Getting the aggregate into Polars took 695 ms here (nearly all of it Spark running the group-by) and the rolling computation after it 5 ms, which is the ratio you'd hope for when Spark is doing the heavy part. (Since polars 0.54, over() returns a Result, hence the ? inside the expression list; nothing I read warned me about that.) The IPC copy is cheap: once to_arrow() has produced the 77.3 MiB IPC file for the whole 1.67-million-row slice, reading it into Polars takes 72 ms.

A Polars frame can keep going into DuckDB too, by the same route the Spark result took. Polars writes the Arrow IPC stream format with IpcStreamWriter, arrow-rs reads it with StreamReader, and the result is a RecordBatch for the same table function; for the 360 rolling rows that is 17.5 KiB and 1 ms:

let mut buf = Vec::new();
IpcStreamWriter::new(&mut buf).finish(&mut rolling.clone())?;
let back: Vec<RecordBatch> = StreamReader::try_new(Cursor::new(buf), None)?.collect::<Result<_, _>>()?;
let _rolling = arrow_view(&db, "rolling", concat_batches(&back[0].schema(), &back)?)?;

After that rolling is a DuckDB view like orders was, and QUALIFY row_number() OVER (PARTITION BY city ORDER BY revenue_7d DESC) = 1 (a clause that filters on a window function's result, which Spark SQL also picked up in 4.2) picks each city's best week out of it.

Step 5: Sending the answer back to Spark

Spark Connect already takes Arrow inbound: createDataFrame is sent as a LocalRelation, the protocol's message for client-side data, whose payload is Arrow IPC bytes (step 6 of the OCaml post was writing exactly that by hand). The Rust client's public entry point for it is create_dataframe(rows, schema), which takes Row values. So a RecordBatch that came out of DuckDB gets unpacked into rows first (a loop over columns and a match on the Arrow type; the demo's record_batch_to_rows is about forty lines) and then written like any other DataFrame:

let (rows, schema) = record_batch_to_rows(&surge_daily)?;
spark.create_dataframe(rows, schema)?
    .write().mode("overwrite").save_as_table("sfo_surge_daily")?;

struct<day:date,orders:bigint,revenue:double,surged_revenue:double,avg_multiplier:double>
  PASS  Spark table rows == DuckDB rows: 30
  PASS  Spark sum(surged_revenue) == DuckDB sum: 103926081.03 vs 103926081.02

That is 157 ms for 30 rows, timed as one step from Arrow to the saved table. (The cent between the sums is floating point, not lost data: Spark and DuckDB add the same thirty doubles in a different order, so the check allows five cents rather than demanding an exact match.) The row conversion grows with row count the same way collect() does in the other direction, so for big data going back, write it somewhere both sides can read and spark.read() it instead. A create_dataframe that took batches directly would be a very welcome addition to the client, since the IPC writer it needs is already in there.

Spark Type Mapping to DuckDB and Polars

The full table with the values is in the demo README; this is the part to keep nearby:

Spark type	arrow-rs	DuckDB	Polars
`tinyint` to `bigint`	`Int8` to `Int64`	`TINYINT` to `BIGINT`	`i8` to `i64`
`float`, `double`	`Float32`, `Float64`	`FLOAT`, `DOUBLE`	`f32`, `f64`
`decimal(p,s)`	`Decimal128(p, s)`	`DECIMAL(p,s)`, digits intact	`decimal[p,s]`, digits intact
`string` (a `char(n)` literal also arrives as `string`)	`Utf8`	`VARCHAR`	`str`
`binary`	`Binary`	`BLOB`	`binary`
`boolean`	`Boolean`	`BOOLEAN`	`bool`
`date`	`Date32`	`DATE`	`date`
`timestamp`	`Timestamp(µs, "<session zone>")`	`TIMESTAMP WITH TIME ZONE`	`datetime[μs, <zone>]`
`timestamp_ntz`	`Timestamp(µs)`	`TIMESTAMP`	`datetime[μs]`
`array<T>`	`List`	`T[]`	`list[T]`
`map<K,V>`	`Map`	`MAP(K, V)`	`list[struct[2]]` (Polars has no map type)
`struct<...>`	`Struct`	`STRUCT(...)`	`struct[n]`
`interval day to second`	`Duration(µs)`	`INTERVAL`	`duration[μs]`
`interval year to month`, calendar `interval`	`Interval`	`INTERVAL`	polars 0.55 panics building the frame (calendar intervals can be imported as a struct with `POLARS_IMPORT_INTERVAL_AS_STRUCT=1`, marked unstable); cast to a number in Spark first
`time`	fails on the server with `UNSUPPORTED_TIME_TYPE`: Spark's Arrow serialiser doesn't handle TIME, so it never leaves
`variant`	`Struct(value Binary, metadata Binary)`	`STRUCT(value BLOB, metadata BLOB)`	`struct[2]`; opaque bytes in both
`void` (a bare `NULL` literal)	`Null`	duckdb-rs can't bind it (`Unsupported data type: Null`)	`null`

For example, timestamp carries the server's session zone into DuckDB's and Polars' type systems (which is correct, and is why the DATE cast in step 3 needed care); timestamp_ntz doesn't. And Polars has no year-month interval type, so turn any such column into a number in Spark before it heads that way. Everything else, decimals included, keeps its precision and lands as a native column type in both engines (in pandas a decimal column becomes a column of Python Decimal objects, exact but slow, which is a thing I'd rather not think about).

Limitations

All of it ran on one machine with the server in a container on the same host, so the Spark-to-client times contain no real network; across a network the transfer would dominate and the gap between collect() and batches would matter less in proportion. Every timing is a single run in a warm process, not a benchmark harness, and the DuckDB view-versus-table gap is an order of magnitude, not a precise ratio. The server ran with default Arrow batch settings. The type sweep didn't include GEOMETRY or GEOGRAPHY, which are new in Spark 4.2. Neither to_datafusion() nor the client's WebAssembly UDF path (Rust functions compiled to wasm and run on the executors) was exercised; the latter is its own blog and probably a fun one. The Polars findings are for the 0.55.2 crate, not Polars 2.0, which Rust can't use yet.

Summarizing the Stack

Server: any Spark 4.2+ Connect server; docker compose up -d --wait with the official image and a health check on the port.
Build: protoc, a C++ toolchain and CMake installed; the Cargo.toml above, copied exactly; rust-toolchain.toml pinned to 1.98.
Connect: SparkSession::builder().remote("sc://host:15002").get_or_create()?, then print spark.version() before anything else.
Get results out as Arrow: to_arrow() for DuckDB 2.0 and Polars (the IPC bytes cross any arrow-rs major), collect_record_batches() for your own arrow-rs 58 code, collect() only for rows you intend to read.
DuckDB: register ArrowVTab, wrap the batch in an ArrowBatchRegistration and keep it alive, put a view over it, and copy it into a table if you'll query it more than once or join it at all.
Polars: IpcReader over the to_arrow() bytes and go; turn on the timezones feature before the first timestamp.
Back to Spark: create_dataframe(rows, schema) for small results; shared storage for large ones.
Types: decide between timestamp and timestamp_ntz before data crosses a boundary, and keep intervals out of Polars.

The idiot's guide said to check that the client you pick supports what you need before you build on it. For the Rust client that check comes out well, and what it's best at is the thing the protocol was designed around in the first place: the result is already Arrow, so the only question left is where you want to send it.

The Ultimate Guide to Apache Spark 4.3

lnc — Mon, 09 Nov 2026 19:52:28 +0000

A release about time (mostly)

Spark 4.3.0 came out on October 30, including the more in depth Databricks release blog, which means my feeds are about to fill with listicles that paste the release notes into a numbered list and call it a day. I wanted to do something a little more in depth by reviewing each of the major features in this release as closely as possible, so that people can get a better idea of the expected behaviours and what it's like to actually run the changes. Funnily, I expect a blog like this to have a shorter shelf life than usual as these releases are currently en track for a quarterly cadence, meaning Spark is about to get much more fully featured in the next year as it moves towards 5.0.

Reading the release notes end to end, I kept noticing that the big items all circle the same thing from different angles. Real-time Mode is about how quickly one record gets through a stateful query; SCD Type 2 is about how long a row's history survives; ASOF JOIN is about which version of a fact was true at a given moment; and nanosecond timestamps are about how finely Spark can tell two moments apart in the first place. The remaining features (Project Feather, the Spark Connect observability work, DataFrame.zip, Python UDF transpilation) are about a different clock, which is yours: how long you wait between writing a query and seeing it run.

Most of Spark 4.3 is about time, both the data's and yours: the scheduler learned to pipeline shuffles, so stateful streaming queries now run at millisecond latency; Auto CDC learned to keep history; SQL learned point-in-time joins and nanosecond precision; and the rest of the release shortens the loop between writing Spark code and finding out whether it works.

Every example here is a runnable script in spark-demos/demos/16_spark_43_guide, and every clip below is that script running against Spark 4.3.0 (pip install pyspark==4.3.0) on my machine. The Real-time Mode clips come from the Real-time Mode demo, which is a teaser for one of my upcoming Spark Streaming blog posts ;)

Below is a rundown of the new features and how to enable or take advantage of them.

Feature	Area	On by default?	How to turn it on
Stateful Real-time Mode	Structured Streaming	Yes (with the Real-time trigger)	`.trigger(realTime="...")`, update mode
SCD Type 2 in Auto CDC	Declarative Pipelines	Yes	`STORED AS SCD TYPE 2` / `stored_as_scd_type=2`; target must support DSv2 row-level ops
Shuffle-free single-task execution	Project Feather	No	`spark.sql.optimizer.singleTaskExecution.enabled=true`
Arrow-native Dataset cache	Project Feather	No	`spark.sql.cache.serializer=...ArrowCachedBatchSerializer` (static)
Operation IDs, REST endpoints, `with` sessions	Spark Connect	Yes	nothing
Persistent local Connect server	Spark Connect	No	`SPARK_LOCAL_CONNECT_REUSE=1`
`DataFrame.zip`	PySpark, Connect	Yes	nothing
Python UDF transpilation	PySpark	No (experimental)	`spark.sql.experimental.optimizer.transpilePyUDFs=true` plus ANSI mode
`ASOF JOIN`	SQL	No (preview)	`spark.sql.join.asofJoin.enabled=true`
`JSON_TABLE`, VARIANT edit functions	SQL	Yes	nothing
Nanosecond timestamps	SQL	No (preview)	`spark.sql.timestampNanosTypes.enabled=true`
`UNNEST`, `counter_diff`, `jaro_winkler_similarity`	SQL	Yes	nothing
Reading inside tar, zip and 7z archives	File sources	No	`spark.sql.files.archive.reader.enabled=true`

Stateful Real-time Mode

Real-time Mode landed in 4.1 as a new trigger for Structured Streaming: instead of slicing a stream into micro-batches and waiting for each one to be planned, scheduled and committed, the query runs long-lived tasks that process records as they arrive and checkpoint on a much longer interval. The catch in 4.1 and 4.2 was that it only ran stateless queries (filters, projections, stream-static joins), because anything stateful needs a shuffle, and a classic shuffle is a blocking barrier where every map task has to finish writing before a single reduce task can read.

A new PipelinedShuffleDependency (SPARK-58185) routes data through a new StreamingShuffleManager (SPARK-57141), which pushes records to downstream tasks over Netty while the upstream tasks are still running. The DAGScheduler now schedules a pipelined group of stages together, fails them together, and refuses to start a group without enough task slots (SPARK-58263, SPARK-58398). This admission check prevents deadlock on a busy cluster: a consumer stage started without its producer would sit idle while holding cores. With that machinery in place, Real-time Mode supports streaming aggregation (SPARK-58635), dropDuplicates (SPARK-58508) and Scala/Java transformWithState (SPARK-57228), and is no longer marked @Experimental.

The query itself doesn't change shape. In this deduplication and windowed-count step from the bookshop demo, only the trigger and output mode are specific to Real-time Mode:

trending = (clicks
    .withWatermark("ts", "10 seconds")
    .dropDuplicates(["event_id"])                       # resent clicks counted once
    .groupBy(F.window("ts", "10 seconds"), "aisle")
    .count())

(trending.writeStream.format("kafka")
    .outputMode("update")
    .trigger(realTime="30 seconds")                     # the whole change
    .start())

Measured end to end on my machine (from the click's own timestamp to the broker's append time on the output topic, 200 clicks a second for 45 seconds), the same query looks like this in each mode:

Step	Mode	p50 ms	p99 ms
Dedup + windowed count	micro-batch	237	645
Dedup + windowed count	Real-time	50	99
Dedup + windowed count	Real-time, flush wait 1 ms	2	41
`transformWithState` recommender	micro-batch	190	396
`transformWithState` recommender	Real-time	29	158

The 50 ms median comes from a timer. The streaming shuffle flushes when its buffer fills or spark.shuffle.streaming.networkBufferMaxWaitTimeMs expires; the default is 50 ms, and a lightly loaded query does not fill the buffer first. Reducing the wait to 1 ms brought the median down to 2 ms at the cost of more small network writes, a tradeoff to tune per workload rather than a useful global default. (The shorter run in the clip measured a 67 ms median and 96 ms p99 at the default, then 2 ms and 3 ms at 1 ms; the table contains the longer benchmark, and this amount of variation is normal for a single-machine test.) Databricks reports p99 latencies of 136 ms for dropDuplicates, 139 ms for a window aggregation and 141 ms for transformWithState on its benchmark workloads (SPARK-58884). Those results are not guarantees, and neither are mine, which come from one desktop with Kafka running in Docker.

For a stateful example, the recommender keeps a taste vector per shopper, built from book embeddings, in transformWithState and rescores the catalog on every click. The recommendations therefore change a few milliseconds after each click:

Real-time Mode still has several limits. transformWithState supports only Scala and Java, requires the RocksDB state store, and fires a processing-time timer on an idle key only when another row reaches its partition or the Real-time batch ends. ASOF JOIN is not supported. Pipelined stages also require cores concurrently, so the cluster must be sized for the whole pipeline rather than its widest stage.

SCD Type 2 in Auto CDC

Auto CDC arrived in 4.2 as the Declarative Pipelines way to turn a change feed into current state: give it keys, a sequencing column and a delete rule, and it keeps a target table in sync with MERGE, handling out-of-order and replayed events for you. In 4.2 that meant SCD Type 1, where the target only ever holds the latest version of each row. 4.3 adds SCD Type 2 (SPARK-58321), in Python (stored_as_scd_type=2, track_history_column_list), in SQL (STORED AS SCD TYPE 2, TRACK HISTORY ON) and through Spark Connect, so the target keeps every version, each bounded by __START_AT and __END_AT.

I reused the ghost kitchen from that post, this time tracking its menu. Each dish change arrives on a Kafka topic as a JSON event, as it would from a CDC tool. The pipeline has a Python source that reads the topic,

@dp.table
def menu_changes():
    return (spark.readStream.format("kafka")
        .option("kafka.bootstrap.servers", "localhost:29292")
        .option("subscribe", "menu-changes")
        .option("startingOffsets", "earliest")
        .load()
        .select(from_json(col("value").cast("string"), CHANGE).alias("e"))
        .select("e.*"))

and this one SQL statement:

-- Keep every version of every dish, not just the latest one.
CREATE STREAMING TABLE menu
FLOW AUTO CDC
FROM STREAM(menu_changes)
KEYS (dish_id)
SEQUENCE BY seq
APPLY AS DELETE WHEN op = 'DELETE'
COLUMNS * EXCEPT (op)
STORED AS SCD TYPE 2
TRACK HISTORY ON (price, station);

TRACK HISTORY ON controls which changes open a new version. Changes to unlisted columns update the current version in place, which prevents corrections such as typo fixes from creating history entries. Publishing change events in batches and running the pipeline after each batch shows both cases:

After the third run, the table contains the full history of each dish:

dish_id	dish	price	station	`__START_AT`	`__END_AT`
1	Birra tacos	14.00	grill	1	2
1	Birria tacos	14.00	plancha	2	NULL
2	Smash burger	13.50	grill	1	2
2	Smash burger	15.00	grill	2	NULL
3	Khao soi	16.00	wok	1	2

The typo fix at sequence 3 did not open a third row for the tacos because dish is not tracked; it updated the current version instead. Removing khao soi closed its window at sequence 2 rather than deleting it, preserving the previous menu state. __START_AT and __END_AT contain the sequencing value, which is an integer here but could instead be a commit timestamp.

As with Type 1, the target must support DSv2 row-level operations because Auto CDC applies each micro-batch with MERGE. Spark's built-in file formats, including Parquet, fail at startup with AUTOCDC_TARGET_DOES_NOT_SUPPORT_MERGE; Apache Iceberg is the usual choice. Iceberg has not published a Spark 4.3 runtime yet (Maven Central currently stops at iceberg-spark-runtime-4.1), so I used an Iceberg runtime built from a branch with Spark 4.3 support. The Iceberg runtime, Kafka connector jars and spark.sql.extensions are static configurations and must be passed to spark-pipelines run, which accepts spark-submit arguments. Putting them in spark-pipeline.yml makes the CLI apply them to a live session and fail with CANNOT_MODIFY_CONFIG.

Project Feather: faster on small data

Spark has always been a slightly absurd tool for a 500-row file because its planning assumes the data is large: a groupBy gets a partial aggregate, a hash shuffle and a final aggregate, while an orderBy adds a range shuffle. Each exchange creates a stage boundary and another scheduling round trip. That overhead disappears into the work on a cluster; on a laptop, in a test suite, or in an agent running hundreds of small queries, it becomes the work. Project Feather (SPARK-56978) addresses that overhead, and 4.3 includes its first two pieces, both opt-in.

Shuffle-free single-task execution (SPARK-57851, spark.sql.optimizer.singleTaskExecution.enabled) notices when a query fragment reads something that fits in one partition (by default a single non-empty small file, or a local relation of up to 1,000 rows) and plans the scans, aggregations, distincts, windows, sorts and limits above it as one task with no exchanges at all. The same aggregation, with the flag off and then on:

def run(single_task, runs=20):
    spark.conf.set("spark.sql.optimizer.singleTaskExecution.enabled",
                   single_task)
    t = time.perf_counter()
    for _ in range(runs):
        # a fresh DataFrame each time, like a notebook cell or a test
        df = (spark.read.parquet(small)
              .groupBy("store").count().orderBy("store"))
        df.collect()
    return df, (time.perf_counter() - t) / runs * 1000

singleTaskExecution	Operators in the executed plan	Time per query
false	Sort, Exchange (range), HashAggregate, Exchange (hash), HashAggregate, FileScan	82.9 ms
true	Sort, HashAggregate, FileScan	51.2 ms

Both exchanges (and the shuffle query stages AQE wraps around them) are gone, and the query takes about 60% of the time, warm, on a session that's already paid for JVM startup. The loop builds a new DataFrame every time on purpose, because that's what a notebook cell, a test or an agent does. My first version collected the same DataFrame twenty times, and from the second collect on AQE reuses the shuffle output it already has, skipping the very stages Feather removes, so the two settings came out within a few milliseconds of each other. I wrote up the full benchmark (more queries, cold starts, and a test suite that went from 12.6 s to 5.2 s) in Benchmark Project Feather on a Fresh DataFrame.

The second piece is Arrow-native Dataset caching (SPARK-57268): set spark.sql.cache.serializer=org.apache.spark.sql.execution.columnar.ArrowCachedBatchSerializer when the application starts (it's static) and df.cache() stores Arrow batches instead of Spark's own columnar format, which pairs nicely with everything Arrow already does between the JVM and Python. Compression is available through spark.sql.execution.arrow.compression.codec, but it defaults to none, so out of the box the Arrow cache is bigger than the default one. Broader in-process local repartitioning was deferred to the next release (SPARK-57399).

Spark Connect: operation IDs and REST endpoints

In my idiot's guide to Connect I complained that debugging a Connect app means correlating a client-side error with a server you can't see, and 4.3 fixes the two halves of that. First, the Python and Scala clients now generate an operation ID before every ExecutePlan request (SPARK-58547) and hand it back on success (df.executionInfo.operation_id) and, more usefully, on failure (e.operation_id on the exception), because a request that fails before its first response never learned a server-generated ID. The server publishes the same ID as a Spark local property, so listeners and executor tasks can see it too.

remote = SparkSession.builder.remote("sc://localhost:15043")
with remote.getOrCreate() as spark:      # new in 4.3: sessions are context managers
    ok = spark.sql("SELECT count(*) AS n FROM range(1000000)")
    ok.collect()
    print("succeeded, operation", ok.executionInfo.operation_id)

    try:
        spark.sql("SELECT revenue FROM range(10)").collect()
    except Exception as e:   # the client knows the ID even on failure
        print("failed,    operation", e.operation_id)

Second, the server's sessions and operations are now in the monitoring REST API (SPARK-57941), next to the SQL and streaming endpoints, where before they only lived in the Connect tab of the UI:

curl -s $API/connect/sessions | jq
curl -s $API/connect/operations | jq '.[] | select(.operationId == env.OP)'

Take the ID the client printed for the failure, and the server's record of that same operation has its state, timings, the SQL execution it ran and the full error:

Because the Connect listener writes these events to the event log, the same endpoints work in the History Server after the application exits. Other Connect changes include the session context manager above (SPARK-58159), a SQLContext wrapper for existing code (SPARK-57021), an experimental SPARK_LOCAL_CONNECT_REUSE=1 setting that keeps a local Connect server alive between runs (SPARK-57787), and broader support in the Connect JDBC driver for complex types, DatabaseMetaData keys and SQLSTATE class 08 connection errors. The INVALID_HANDLE.SESSION_* errors changed SQLSTATE from HY000 to 08003, so retry logic that matches SQLSTATE may need updating. Connect Gateway and the Rust-backed Python client are concurrent projects, not Spark 4.3 features.

PySpark: DataFrame.zip and UDF transpilation

DataFrame.zip (SPARK-55886, with Connect support in SPARK-57247) places two column projections of the same DataFrame side by side, row for row, without a join key. It replaces joins built on monotonically_increasing_id(), whose rows are not guaranteed to align. Both sides must originate from the same plan, with matching filters, sorts, joins and aggregations beneath the projections.

The experimental Python UDF transpiler (SPARK-55206) is the one I've been waiting for since the UDF showdown, where the conclusion was that the fastest Python UDF is the one that never runs Python. With spark.sql.experimental.optimizer.transpilePyUDFs=true and ANSI mode on, PySpark reads simple UDF bodies when you define them and lowers them to Catalyst expressions, so len(dish) becomes Spark's own length and the BatchEvalPython step (and the Python worker, and the serialization) drops out of the plan:

spark = (SparkSession.builder
         .config("spark.sql.ansi.enabled", "true")  # transpiling needs ANSI
         .getOrCreate())

@udf("int", useArrow=False)
def name_length(dish: str) -> int:
    return len(dish)

The restrictions are not all documented together, so I read pyspark/sql/udf.py to collect them. Transpilation applies only to non-Arrow batched UDFs (hence useArrow=False), classic PySpark, ANSI mode, and function bodies that PySpark can recover from source, excluding an inline lambda passed directly to udf(). With ANSI mode disabled, PySpark warns and falls back because the generated expressions use ANSI semantics. The transpiler also preserves Python behavior where Spark differs: a transpiled len raises on NULL as CPython would, represented in the full plan by a raise_error branch. Parameter type hints keep the generated plan small. Other PySpark changes bound UDF input batches by bytes, implement more pandas API on Spark functions natively, remove Python 3.10 support, and raise the minimum NumPy version to 1.23.2.

Spark SQL: ASOF JOIN, JSON_TABLE, VARIANT edits and nanoseconds

ASOF JOIN (SPARK-58092) is the point-in-time join: for each row on the left, the single closest row on the right that satisfies a MATCH_CONDITION comparison, optionally within an equality ON or USING key. In my Rust client post I had to pull a Spark result into DuckDB to do exactly this, so it's nice to have it native, with a dedicated sort-merge operator. The canonical ML case is matching each prediction to the feature version that had actually been published when the prediction was made:

SELECT p.made_at, p.score, f.version AS features_used, f.published_at
FROM predictions p
ASOF JOIN feature_versions f
MATCH_CONDITION (p.made_at >= f.published_at)  -- latest one out by then

The 10:03 prediction gets v1 from 10:00 and never v2 from 10:05, which didn't exist yet (using it would be leakage), and the 10:07 prediction gets v2 alone, where a plain >= join would return both versions and leave you to pick the latest yourself, which is exactly where the leakage bugs usually creep in. It's a preview: spark.sql.join.asofJoin.enabled is off by default (and internal, so it won't show up in SET -v), only INNER and LEFT OUTER are supported, and in streaming only stream-static works, with the stream on the left.

JSON_TABLE (SPARK-58366) is the SQL-standard table function for expanding a JSON array into typed rows. It also supports FOR ORDINALITY and EXISTS columns:

And VARIANT values can finally be edited in place with variant_set, variant_insert, variant_delete, variant_array_append and variant_strip_nulls (plus try_ versions), instead of being parsed into a struct and serialized back:

Nanosecond timestamps (SPARK-56876) add TIMESTAMP_NTZ(p) and TIMESTAMP_LTZ(p) for precisions 7 to 9, with unix_nanos, timestamp_nanos, current_timestamp(p), datetime functions, interval arithmetic, and Parquet, ORC, JSON, CSV, Avro, JDBC, Arrow and Connect support. Spark's TIMESTAMP has been microseconds since forever, which means two market events a nanosecond apart (or two rows from a Parquet file written by something that does use nanoseconds) silently became the same instant:

Nanosecond timestamps remain a preview behind spark.sql.timestampNanosTypes.enabled, and the Databricks release post warns that they may change incompatibly. Typed literals such as TIMESTAMP_NTZ(9) '...' do not parse yet, so values must be cast from strings. I would not use the type for production work yet: while building this demo, I found a query that returns an incorrect count when a nanosecond column appears beside other DISTINCT aggregates in the same SELECT; I am reducing the reproduction before filing it. The TIME type also gained nanosecond precision and broader connector support, but remains disabled behind an internal flag.

Other SQL additions include ANSI UNNEST in FROM, with WITH ORDINALITY; counter_diff, a window function that converts a running counter such as requests served or bytes sent into per-interval deltas and returns NULL after a reset; and jaro_winkler_similarity for fuzzy name matching:

Also new and not filmed: @-syntax time travel on table names (t@v3, or t@20261105093000000 in the fixed 17-digit yyyyMMddHHmmssSSS form), COMMENT ON COLUMN, dynamic table options on UPDATE, MERGE and DELETE, AS JSON output for SHOW TABLES and SHOW PARTITIONS, collect_union, to_base32/from_base32, and a BIN BY relation operator behind a flag.

Reading files inside archives

The archive reader may be the feature I use most. With spark.sql.files.archive.reader.enabled=true, file sources read entries inside tar, zip and 7z archives as separate files, including during schema inference; archivePathFilter selects the entries to read. This removes the usual step of unpacking a vendor's nightly archive into a temporary directory:

Other changes in Spark 4.3

A lot of 4.3 is engine work you get for free. ReplaceHashWithSortAgg and combining adjacent partial and final aggregates are now on by default, whole-stage codegen covers UNION, AQE can convert a sort-merge join to a shuffled hash join in the physical plan, nested column pruning reaches into transform over arrays of structs, and the vectorized Parquet reader got faster decoding paths and a BYTE_STREAM_SPLIT reader. There's an opt-in to spread NULL outer join keys across partitions (a classic skew source), fractional spark.task.cpus, spark.rdd.compress on by default, and the Spark UI's Content Security Policy on by default. On Kubernetes the official images move to Ubuntu 26.04 with Python 3.14, and containers drop allowPrivilegeEscalation by default.

A few changes can change results or break things on upgrade, all in the migration guide: exact percentile/median may differ in the last ULP; Dataset.observe metrics now count only the last successful task attempt; HAVING is applied before window functions when the SELECT also has a generator (fixing wrong results); non-deterministic filters are no longer pushed to DSv2 sources; Thrift Server sessions can no longer set JVM system properties; and the AQE rule DynamicJoinSelection was renamed DemoteBroadcastHashJoin, so an exclusion by the old name now silently does nothing.

In Summary

Stateful Real-time Mode: pipelined shuffles let dedup, aggregation and transformWithState (Scala/Java) run at millisecond latency with only a trigger change, and the default median is a 50 ms flush timer you can tune.
SCD Type 2 in Auto CDC keeps every version with __START_AT/__END_AT; TRACK HISTORY ON decides what opens a new version, and the target needs a row-level-capable catalog like Iceberg (whose 4.3 runtime isn't published yet).
Project Feather is opt-in: single-task execution removes the exchanges from small queries (about 1.6x faster here, warm), and the Arrow cache serializer is a static config.
Spark Connect gives you an operation ID on success and failure, and the same operation in the server's REST API.
DataFrame.zip replaces the monotonically_increasing_id join hack; UDF transpilation removes Python from simple UDFs if you're on classic PySpark with ANSI on.
ASOF JOIN, nanosecond timestamps and archive reading are all behind flags; JSON_TABLE, VARIANT edits, UNNEST and counter_diff just work.

Most of it comes back to time: a record's few milliseconds through a stateful query, a dish's price history that never gets overwritten, the feature version that was true at 10:03, the nanosecond between two trades, and the milliseconds a small query no longer spends on shuffles. On a quarterly cadence, expect all of it to keep moving faster than it used to.

Building a Spark Connect Client in OCaml because I'm crazy and you're not

lnc — Tue, 20 Oct 2026 22:18:03 +0000

But Lisa, surely you must have a good reason for using OCaml?

No, not really. Well, yes. Sort of.

In a previous blog post I wrote a cheeky little guide to Spark Connect Clients, which I enjoyed writing thoroughly and was pretty well received. But it ended on a warning: not all Connect Clients are built equally, and it highly depends on the language you're using and the love and care of the maintainers. Current gold standards include the Rust Client, for instance, which gives you full interoperability with Rust native engines like Polars or DuckDB. More on that in a follow up post, but it's pretty cool.

The follow-up question is what it takes to build one then and how to test it. And to be honest, I've always defaulted to OCaml as sort of a joke language- partially from it's niche quality and mostly from my secret love for the finance industry and beasts like Jane Street.

OCaml is a useful language for the demonstration because it lacks all of the built in capabilities that would otherwise make life easy: there is no Arrow library, no concept of the gRPC protocol at all, and a protobuf compiler is its own project with its own parser rather than a plugin for Google's protoc. Every piece of the Connect client has to be manually built; making you realize how good the Go or Rust clients have it in comparison because of their respectively rich ecosystems. Naturally, I vibe coded most of this with my good friend Claude Code, and if you want to skip the entire droll on the implementation journey you can skip ahead to the repo here.

Every Spark Connect client essentially has the same four basic requirements: compile the published protos, make a streaming gRPC call, build relation trees, and decode Arrow batches. The Connect server itself does so much of the work for you in this architecture that the only parts costing real effort are the two the protocol leaves to the client implementation: its protobuf toolchain and its Arrow codec.

The Roles and Responsibilities of a Connect Client

The client sits between the DataFrame calls and the server, and it never processes data itself. It turns method calls into a protobuf tree, sends the tree, and turns the Arrow that comes back into values:

The analyzer, Catalyst and the executors are the server's. The rest is the client's.

Step 1: Getting the protos to compile

The protocol is its protobuf definitions, published in the Spark repository under sql/connect/common/src/main/protobuf/spark/connect/. A client starts by copying them out of the tag it targets (I used v4.2.0) and compiling them with whatever protobuf toolchain its language has. They import google/protobuf/any.proto and timestamp.proto, and Spark's error details also need google/rpc/status.proto and error_details.proto. None of these are in the Spark repo. If the language's toolchain does not ship them, each is a few lines to write by hand, and wire-compatible as long as the field numbers match.

This step is "mechanical" in Go, Rust, Swift and Java, where protoc and a maintained plugin do it. In OCaml the compiler is ocaml-protoc, a from-scratch parser and generator, and Spark's protos hit several things it disagrees with. Each fix is one line in a post-generation fix-up script:

Problem	Where	Fix
Single-quoted strings (`syntax = 'proto3';`), legal but rare	every Spark proto	rewrite to double quotes when vendoring
`reserved "userId", "sessionId";` (field name reservations)	`relations.proto`	drop the lines; they have no wire effect
A one-line default emitted before the function it calls exists	the mutually recursive ML relation messages	move the line below its dependency
Key and payload emitted in the wrong order for a oneof case whose message is empty	`Catalog.CurrentDatabase`, `CurrentCatalog`, `ClearCache`	swap the two generated lines

The oneof bug surfaced the way every protobuf mistake surfaces. Three catalog calls, and only those three, came back from the server as UNKNOWN: Application error processing RPC, with Invalid protobuf byte sequence in the server log. That message means the server could not parse the request at all, and the diagnostic is to dump the bytes:

Catalog{current_database}  generated:  00 0a    length 0, then the key for field 1
Catalog{current_database}  correct:    0a 00    key for field 1, then length 0
AnalyzePlan{spark_version} generated:  52 00    same shape, inlined elsewhere, comes out right

ocaml-protoc encodes back to front (payload first, then its key, so a nested message's length is known before the key is written), and the standalone encoder for an empty message had the two lines in the wrong order. Nothing else in the API was affected: the bug lives in the encoder generated for a message that is nothing but a oneof, and Catalog is the only such message on the request path with empty cases. A client in Go never meets this, because protoc has been exercised by every Google service for a decade; a client in OCaml meets it immediately. The protobuf compiler is the first dependency the protocol pushes onto the host language, and the Spark protos are a stress test for it.

Step 2: Speak gRPC

gRPC is small once HTTP/2 is available, and OCaml does have HTTP/2 (h2, with TLS through ocaml-tls and the system trust store). A call is a POST to /spark.connect.SparkConnectService/<Method> with two headers that matter, content-type: application/grpc and te: trailers, plus authorization: Bearer <token> when a token is set. The body in each direction is a sequence of messages, each framed as a compression flag byte, a big-endian length, and the protobuf bytes:

let frame payload =
  let n = String.length payload in
  let b = Bytes.create (5 + n) in
  Bytes.set b 0 '\000';                      (* not compressed *)
  Bytes.set_int32_be b 1 (Int32.of_int n);   (* length *)
  Bytes.blit_string payload 0 b 5 n;
  Bytes.unsafe_to_string b

The status comes back in HTTP/2 trailers, not in the response status line (which is always 200), as grpc-status and a percent-encoded grpc-message. Spark adds grpc-status-details-bin, base64 of a google.rpc.Status, which is where the useful error information lives. The reader side is the mirror of frame: buffer the body, peel off messages as their lengths complete, and finish when the trailers arrive.

let status_of_headers (h : H2.Headers.t) =
  match H2.Headers.get h "grpc-status" with
  | None -> None
  | Some code ->
      let code = try int_of_string code with _ -> 2 in
      let message = Option.value ~default:"" (Option.map percent_decode (H2.Headers.get h "grpc-message")) in
      let details = Option.map base64_decode (H2.Headers.get h "grpc-status-details-bin") in
      Some { code; message; details }

Spark Connect only uses two call shapes: unary (AnalyzePlan, Config, Interrupt, ReleaseSession, FetchErrorDetails, and friends) and server-streaming (ExecutePlan, ReattachExecute). A unary call is a streaming call where the client reads one message. There are no client-streaming or bidirectional methods except AddArtifacts, which is only needed to upload files to the server. That is the whole transport. It worked on the first query, and almost none of the effort in this project went into it.

Step 3: A session, and the first RPC

Every request carries the same identity block: a session_id the client mints (a UUID), a user_context with a user_id, and a client_type string that is effectively a user agent. After the first reply the client also records the server_side_session_id the server sent and echoes it back as client_observed_server_side_session_id on every later request, which is how the server notices if it has restarted underneath the client.

The hello world of Connect is AnalyzePlan with the spark_version option, because it needs no plan:

let analyze t analyze =
  let req =
    P.Base.make_analyze_plan_request ~session_id:t.session_id
      ?client_observed_server_side_session_id:t.server_side_session_id
      ~user_context:(user_context t) ~client_type:t.client_type ~analyze ()
  in
  unary t ~method_:"AnalyzePlan" ~encode:P.Base.encode_pb_analyze_plan_request
    ~decode:P.Base.decode_pb_analyze_plan_response req
  >|= fun r -> observe t r.server_side_session_id; r

let version t =
  Client.analyze t.client B.Spark_version >|= fun r ->
  match r.result with Some (B.Spark_version v) -> v.version | _ -> failwith "unexpected response"

server version: 4.2.0

Once that prints, the protobuf toolchain, the HTTP/2 stack, the gRPC framing and the session plumbing are all known good, and everything after it is vocabulary.

Step 4: Build plans instead of running them

Plan building is the most clerical step of the seven. A Relation is a common block holding a plan_id plus a oneof of every operator the server understands: Read, Project, Filter, Join, Aggregate, Sort, Limit, SetOp, Sample, Deduplicate, Range, Sql, LocalRelation, and a long tail. A DataFrame method is a function that wraps the current relation in a new one, and that is the entire DataFrame API:

let of_rel_type client rel_type =
  { client;
    rel = R.make_relation ~common:(R.make_relation_common ~plan_id:(Client.fresh_plan_id client) ()) ~rel_type () }

let filter t cond = of_rel_type t.client (R.Filter (R.make_filter ~input:t.rel ~condition:cond ()))

let join ?(how = `Inner) t other ~on =
  of_rel_type t.client
    (R.Join (R.make_join ~left:t.rel ~right:other.rel ~join_condition:on ~join_type:(join_type_pb how) ()))

Expressions are the same idea one level down. A column reference is an UnresolvedAttribute holding a string; a literal is a Literal with a typed oneof; and every function and every operator is an UnresolvedFunction called by name, so a + b is the function + applied to two arguments and a == b is ==. The client never resolves anything:

let fn ?(distinct = false) name args =
  of_expr_type (E.Unresolved_function
    (E.make_expression_unresolved_function ~function_name:name ~arguments:args ~is_distinct:distinct ()))

let add a b = fn "+" [ a; b ]
let eq a b = fn "==" [ a; b ]
let gt a b = fn ">" [ a; b ]

There is no analyzer in the client: Column.col "nope" is a string in a message until the server sees the plan, which is why a typo fails late under Connect. The client cannot check the name; it has no catalog to check it against.

The size of this layer depends on how much the server does, and the server does nearly everything. Each of the following is a server message rather than client code:

What you'd expect to implement client-side	What Connect actually does
Rendering the table for `show()`	`ShowString` is a relation; the server formats the ASCII table and returns it as one string cell
`count()`	an `Aggregate` with `count(1)`; the client reads back one row
Schema, `printSchema()`, `explain()`	`AnalyzePlan` with `Schema`, `TreeString` or `Explain`; the server returns the type tree, the formatted text, or the plan string (the only way to see one, since Catalyst isn't on your machine)
`listTables()`, `currentDatabase()`, `tableExists()`	relations under the `Catalog` message; they execute like a query and return rows
`sql()` for DDL vs for a query	a `SqlCommand`; the server runs it eagerly and, for a query, hands back a relation to keep building on
Named and positional SQL parameters	literals attached to the `SqlCommand`; the server binds `:name` and `?`
Telling `a.id` from `b.id` in a self-join	every relation carries a client-assigned `plan_id`, and an attribute can name the plan it belongs to

The plan_id is the one piece of real design in this step. Because the client assigns plan_ids, it can mint a column reference bound to a particular DataFrame, and the server resolves it against that side of the join even when both sides have the same column names:

let col t name = Column.col_of_plan ~plan_id:(plan_id t) name

let a = Dataframe.alias people "a" and b = Dataframe.alias people "b" in
Dataframe.join a b ~on:Column.O.(Dataframe.col a "id" = Dataframe.col b "id" + Column.int 1)

Step 5: Execute, and decode the Arrow that comes back

ExecutePlan takes a Plan (either a Relation to query or a Command to run) and returns a stream of ExecutePlanResponse messages. Each one has a response_type oneof: arrow_batch carries data, sql_command_result carries the relation a SqlCommand produced, result_complete is the end marker, and progress and metrics messages can be ignored at first. The collect loop is a fold over the stream:

let collect_arrow t =
  Client.execute t.client (root t) >>= fun stream ->
  let schema = ref None and batches = ref [] in
  Lwt_stream.iter
    (fun (r : B.execute_plan_response) ->
      match r.response_type with
      | Some (B.Arrow_batch ab) ->
          let sc, bs = Arrow_ipc.read_stream (Bytes.unsafe_to_string ab.data) in
          (match sc with Some s when !schema = None -> schema := Some s | _ -> ());
          batches := List.rev_append bs !batches
      | _ -> ())
    stream
  >|= fun () -> (!schema, List.rev !batches)

Each arrow_batch in ab.data is a complete Arrow IPC stream, a schema message followed by one record batch, not a bare batch sharing a schema across the response. So the decoder is a full IPC stream reader. In a language with an Arrow library this step is one function call. OCaml does not have one (the package on opam binds the C++ library, which defeats the point of a thin client), so this is where the project stopped being clerical.

An IPC stream is a sequence of messages, each framed as a continuation marker, a metadata length, a FlatBuffers Message, and a body of raw buffers:

FF FF FF FF  <int32 metadata length>  <FlatBuffer: Message{version, header type, header, bodyLength}>  <body>

The Message header is a Schema (a vector of Fields, each with a name, a nullability flag, a type union and child fields) or a RecordBatch (a row count, a vector of field nodes giving each column's length and null count, and a vector of buffers giving offsets into the body). Decoding a batch is a depth-first walk: for each field, take the next node, take its validity bitmap buffer, take its data buffer(s), and recurse into children for lists, structs and maps. FlatBuffers itself is a vtable-indirected binary format and reading it is a page or so of offset arithmetic. Then there is one decoder per type. The decoder is the largest file in the project not because Arrow is complicated but because Spark produces many Arrow types:

Spark type	What arrives in the Arrow batch
tinyint, smallint, int, bigint	`Int` with bit width and signedness; a fixed-width buffer
float, double	`FloatingPoint` with precision
string, binary	`Utf8` / `Binary`: an offsets buffer plus a data buffer (and `Large` variants with 64-bit offsets)
decimal(p, s)	`Decimal`: 128-bit (or 256-bit) little-endian two's complement, which needs a bignum-to-decimal-string conversion
date	`Date32`, days since the epoch
timestamp	`Timestamp` in microseconds, with the session time zone as a string
timestamp_ntz	the same with no time zone
interval year to month	`Interval(YEAR_MONTH)`
interval day to second	`Duration(MICROSECOND)`
array, struct, map	`List`, `Struct`, `Map`: child columns with their own nodes and validity bitmaps

Decimal decoding was the only algorithmic piece, and only because the natural answer (the zarith bignum library) is a C dependency I did not want in a client whose point is having none. Performance without tuning was acceptable: a two-column projection over three million rows (a bigint and a SHA-1 string) came back and decoded in 1.63 seconds on localhost, with the client building a boxed value per cell, the least efficient representation available. The server's time dominates.

Step 6: Sending data to the server

createDataFrame has to send rows to the server, and Connect's LocalRelation message carries them as, again, an Arrow IPC stream. A client that builds a DataFrame from local values therefore needs an Arrow writer, including a FlatBuffers builder for the schema and record batch metadata:

let create_dataframe ?schema t ~columns rows =
  let data = Arrow_writer.encode ?schema ~columns rows in
  of_rel_type t (R.Local_relation (R.make_local_relation ~data:(Bytes.unsafe_of_string data) ()))

The builder is the fiddly half. FlatBuffers are written back to front with vtables, alignment padding and forward offsets, and a byte out of place produces a Java stack trace on the server rather than anything that points at the byte. So the acceptance test for the writer was not the Spark server, which is forgiving, but pyarrow's validate(full=True), which is not. The stream passed with every Spark type the decoder reads, including nested lists, structs and decimals. The server infers the Spark schema from the Arrow schema, so a local table of Value.Int, Value.String and Value.Float cells comes up as bigint, string and double with no DDL string needed (a DDL string can be passed for narrower types).

Every official client carries both halves of this codec; they borrow them from an Arrow library the language already had.

Step 7: Errors, the way Spark sends them

The grpc-status-details-bin trailer is base64 of a google.rpc.Status, whose details is a list of Any. Spark puts a google.rpc.ErrorInfo in it, and the metadata map on that is where everything useful lives: classes (the JVM exception class chain as a JSON array), sqlState, errorId, and messageParameters. The error id goes to FetchErrorDetails for the server-side stack trace and query context.

let status = P.Status.decode_pb_status (Pbrt.Decoder.of_string raw) in
let info =
  List.find_map
    (fun (a : P.Any.any) ->
      if ends_with ~suffix:"google.rpc.ErrorInfo" a.type_url then
        Some (P.Error_details.decode_pb_error_info (Pbrt.Decoder.of_string (Bytes.to_string a.value)))
      else None)
    status.details
in
(* info.metadata: ("classes", "[...]"), ("sqlState", "42P01"), ("errorId", "..."), ... *)

The Spark error condition itself is not a separate field; it is the bracketed prefix of the message, so a client lifts it with a trivial parse. The OCaml client's output for a query on a table that does not exist, with no string matching beyond that bracket:

Spark error INTERNAL (org.apache.spark.sql.catalyst.ExtendedAnalysisException):
[TABLE_OR_VIEW_NOT_FOUND] The table or view `nope` cannot be found. ...
SQLSTATE: 42P01; line 1 pos 14;
'Project [*]
+- 'UnresolvedRelation [nope], [], false

error_class = TABLE_OR_VIEW_NOT_FOUND, sql_state = 42P01, and the exception classes as a list. The server supplies all of it; the client's job is to keep it.

Putting it together

With those seven pieces the client reads like every other Connect client, because they all converge on the proto's shape:

open Lwt.Infix
open Spark_connect

let () =
  Lwt_main.run begin
    Session.connect "sc://localhost:15002" >>= fun spark ->
    let people = Session.create_dataframe spark ~columns:[ "id"; "name" ]
      [ [ Value.Int 1; Value.String "ann" ]; [ Value.Int 2; Value.String "bob" ] ] in
    let orders = Session.create_dataframe spark ~columns:[ "pid"; "amount" ]
      [ [ Value.Int 1; Value.Float 10.0 ]; [ Value.Int 1; Value.Float 5.5 ] ] in
    let per_person =
      Dataframe.join people orders ~on:Column.O.(Column.col "id" = Column.col "pid")
      |> fun df -> Dataframe.group_by df [ Column.col "name" ]
      |> fun g -> Dataframe.Grouped.agg g [ Column.alias (F.sum (Column.col "amount")) "total" ] in
    Dataframe.show per_person >>= fun () ->
    Session.close spark
  end

+----+-----+
|name|total|
+----+-----+
| ann| 15.5|
+----+-----+

The two local tables went up inside the plan as LocalRelations, the join and aggregate went up as relation nodes around them, the server formatted the table and returned it as a one-cell Arrow batch, and the whole exchange was a single ExecutePlan call, since nothing before show touches the server. Pointed at a Databricks endpoint with sc://host:443/;token=..., the same client turns TLS on because the token is present, which is what the connection string spec asks for and what every client reimplements.

The SQL path is the only one with a wrinkle: sql() sends a SqlCommand and executes eagerly, so CREATE TABLE runs right away, and for a query the reply carries a sql_command_result relation that the client wraps as the DataFrame to keep building on. Config is a unary call with get/set/unset operations, the catalog is relations, writers are WriteOperation (v1, with save modes) and WriteOperationV2 (create/replace/append) commands, and persist is an AnalyzePlan option.

Testing its Production Readiness

So, here begets the real question: is this actually usable? Well, maybe. As much as I'd love Jane Street to plug this in and hire me to maintain it (though, I'm pretty happy at Databricks right now), there's still some coverage issues, particularly around UDFs which are highly variable anyways.

For my own due diligence, I tested the client two ways: every message it can build, exercised against more than one Spark release, and execution on a real cluster of the kind of work I naively imagine someone at Jane Street would reach for OCaml to do.

Every message, tested against three Spark releases

Coverage is precisely measurable for a Connect client, because the protocol is a finite list of protobuf messages. The repo has a script that reads the generated bindings, finds which relations, expressions, commands, catalog calls and analyze requests the client builds, follows each one back to the public function that builds it, and checks whether a test calls that function. Its output is docs/protocol-coverage.md, regenerated rather than maintained by hand:

Protocol area	Implemented	Of those, tested
Relations (DataFrame operations)	33 of 61	all
Expressions	14 of 22	all
Literal types	18 of 22	all
Catalog	21 of 36	all
AnalyzePlan requests	11 of 14	all
Config operations	7 of 7	all
Commands	4 of 20	all

The same suite passes against Spark 4.1.3, 4.2.0 and 4.3.0-rc1. The unimplemented messages are mostly the runtime half of the protocol (streaming, UDFs, ML, pipelines), plus a tail of statistics relations like crosstab and approx_quantile that I have not needed yet.

The tests found bugs that a demo doesn't make obvious. na_fill sent whole numbers as 32-bit integer literals, which the server's fill rejects outright (PySpark always sends longs, so it never meets this). Sort orders without an explicit null ordering went up as "unspecified", so an ascending sort put nulls last where Spark and PySpark put them first. And show() did not end with a newline, so two tables printed in a row ran together. The suite also had a bug of its own: it compared a TIMESTAMP literal without pinning the session time zone, so it passed on every server running in UTC (all of my Docker ones) and failed on the one running natively on my machine.

Case Study: A small options desk

OCaml's best-known home outside academia is finance, where trading firms write their pricing and risk code in it, wirh the same firms keep their market history in data lakes that Spark reads. OCaml and Spark here have historically met through Python or a SQL string.

So my E2E is a small options desk with one question: should 30-day index options be priced off realized volatility, measured from recent prices, or off the VIX, the market's implied volatility?

The pricing code is borrowed. It is the closed-form Black-Scholes module from Case Ollila's option-pricing-model-comparison (MIT), typed OCaml with Greeks in trader units and every numeric routine written from scratch, vendored unchanged into the example. The market data is public: daily S&P 500, Nasdaq Composite and Dow Jones closes, the VIX, and the 3-month Treasury yield, all from FRED. The whole thing runs as one OCaml program, spark-demos/demos/08_spark_connect_ocaml, against a standalone Spark 4.2 cluster with Delta Lake:

The program is a pipeline, and each step that can check itself does so against an independent calculation:

1. Market data through a Python adapter. The adapter is a short Python data source that downloads FRED series, one partition per series, installed on every node as a pyspark_fred package so Spark discovers it with no per-session registration. From OCaml the adapter is a format name:

let fred spark series =
  Session.read spark |> fun r ->
  Session.Reader.format r "fred" |> fun r ->
  Session.Reader.option r "series" (String.concat "," series) |> fun r -> Session.Reader.load r

The client never learns that Python is involved: it sends a Read relation naming a format, the executors run the adapter, and the same Arrow batches come back that the decoder reads for any query. The ingest lands ten years of closes in a Delta table (ten years because that is what FRED's agreement with S&P Dow Jones Indices lets it publish for their series, and the job trims the Nasdaq to the same window):

+---------+--------+----------+----------+
|   series|count(1)| min(date)| max(date)|
+---------+--------+----------+----------+
|     DJIA|    2514|2016-10-10|2026-10-09|
|NASDAQCOM|    2515|2016-10-10|2026-10-09|
|    SP500|    2514|2016-10-10|2026-10-09|
+---------+--------+----------+----------+

2. The heavy part, in Spark. Daily log returns come from a lag over each series, realized volatility is the standard deviation of the last 21 returns annualized by the square root of 252 trading days, and drawdown is each close against the running maximum, all window functions built in OCaml and evaluated on the cluster:

let by_day = Column.Window.partition_by [ col "series" ] |> fun w -> Column.Window.with_order_by w [ col "date" ] in
let last_21 = Column.Window.rows_between by_day (Column.Window.Value (Column.int (-20))) Column.Window.current_row in
let daily =
  Dataframe.with_columns prices [ ("ret", F.log Column.O.(col "value" / Column.over (F.lag (col "value")) by_day)) ]
  |> fun d -> Dataframe.with_column d "rv21" Column.O.(Column.over (F.stddev (col "ret")) last_21 * F.sqrt (Column.int 252))

+---------+----------+--------+--------+------------+----------------+-------------+
|   series|      date|    spot|rv21_pct|avg_rv21_pct|max_drawdown_pct|worst_day_pct|
+---------+----------+--------+--------+------------+----------------+-------------+
|     DJIA|2026-10-09|51654.95|    10.2|        14.3|           -37.1|        -13.8|
|NASDAQCOM|2026-10-09|27366.17|    14.2|        19.4|           -36.4|        -13.1|
|    SP500|2026-10-09| 7811.54|    10.0|        15.0|           -33.9|        -12.8|
+---------+----------+--------+--------+------------+----------------+-------------+

The worst single day for all three is 16 March 2020. The deepest drawdown is March 2020 for the S&P 500 and the Dow, but December 2022 for the Nasdaq, whose 2022 bear market went further than the Covid crash. Joining the S&P 500's trailing realized volatility to the VIX day by day answers half the desk's question: over the decade the VIX sat above realized volatility on 86.1% of trading days, by 3.7 volatility points on average. That gap is a rough measure of the volatility risk premium, the price the market charges for insurance.

3. Pricing, in OCaml. The program pulls the latest close and realized volatility for each index, plus the current VIX and T-bill yield, and hands them to the pricer as its author intended (I set the dividend yield to zero to keep the example small):

let m = { T.spot; strike = spot; ttm = 30. /. 365.; rate; div_yield = 0.0; vol = rv series } in
let g = BS.greeks kind m in
(BS.price kind m, g.delta, BS.vega_per_point g.vega, BS.theta_per_day g.theta)

  risk-free (3m T-bill) 4.23%, VIX 15.4
  index      kind contracts     strike    vol  premium   delta   vega/pt theta/day
  SP500      call        20    7811.54  10.0%   103.40   0.554      8.85     -1.96
  SP500      put        -10    7811.54  10.0%    76.28  -0.446      8.85     -1.06
  NASDAQCOM  call         5   27366.17  14.2%   492.76   0.542     31.13     -9.03
  DJIA       put          8   51654.95  10.2%   516.15  -0.447     58.56     -7.22
  SP500 30-day ATM call: 103.40 at realized vol (10.0%), 151.41 at VIX (15.4%)

The same at-the-money S&P call is worth 103.40 at realized volatility and 151.41 at the VIX, 46% more, which is the volatility risk premium expressed in dollars. Put-call parity, the pricer's own check that its calls and puts are consistent with each other, holds for every index to within 5e-13. The priced book then goes back to Spark the other way, as a LocalRelation built from OCaml values and written as a Delta table.

4. The risk layer, as a metric view. Spark 4.2 added metric views (SPARK-54403, SPARK-54405): a view defined in YAML as a source, dimensions, and measures, where each measure is an aggregate defined once and queried with measure() under whatever grouping the query asks for. A risk layer has the same shape, so the desk's exposures are defined once:

version: "0.1"
source: desk_book
dimensions:
  - name: index_name
    expr: index_name
  - name: kind
    expr: kind
measures:
  - name: premium_usd
    expr: cast(round(sum(premium * contracts * 100)) as bigint)
  - name: dollar_delta
    expr: cast(round(sum(delta * contracts * 100 * spot)) as bigint)
  - name: vega_usd_per_pt
    expr: cast(round(sum(vega_pt * contracts * 100)) as bigint)
  - name: theta_usd_per_day
    expr: cast(round(sum(theta_day * contracts * 100)) as bigint)

The client has no code for metric views at all. Defining one is a SQL statement, and querying one from the DataFrame API is a function call, because measure is a function the server resolves:

Dataframe.group_by (Session.table spark "desk_risk") [ col "index_name" ] |> fun g ->
Dataframe.Grouped.agg g (List.map (fun m -> Column.alias (Column.fn "measure" [ col m ]) m)
                           [ "premium_usd"; "dollar_delta"; "vega_usd_per_pt"; "theta_usd_per_day" ])

+----------+-----------+------------+---------------+-----------------+
|index_name|premium_usd|dollar_delta|vega_usd_per_pt|theta_usd_per_day|
+----------+-----------+------------+---------------+-----------------+
|      DJIA|     412920|   -18467411|          46844|            -5775|
| NASDAQCOM|     246379|     7417769|          15563|            -4514|
|     SP500|     130507|    12138611|           8853|            -2867|
+----------+-----------+------------+---------------+-----------------+

The check computes the same four sums directly from the book table and compares them row by row. They match.

5. The next day. The desk adds to its S&P calls and closes the Nasdaq position, as a MERGE from a DataFrame of fills built in OCaml. Delta keeps the previous book, and VERSION AS OF reads it back:

  yesterday: [DJIA, put, 8] [NASDAQCOM, call, 5] [SP500, call, 20] [SP500, put, -10]
  today:     [DJIA, put, 8] [SP500, call, 25] [SP500, put, -10]

Check	Compared against	Result
Put-call parity, each index	the pricer's own arbitrage identity	pass
Priced book written to Delta	the number of positions sent	pass
Risk measures from the metric view	the same aggregation on the book table	pass
Time travel to yesterday's book	the book before the `MERGE`	pass

On the cluster, every Spark job succeeded, with the tasks split close to evenly between the two workers' executors.

What's left on my to do list, and where the official clients spend their time

What this client does not do is also a map of where Connect stops being a plan-and-Arrow protocol and starts being a runtime:

Reattachable execution. The official clients use ReattachExecute and ReleaseExecute so a dropped connection can resume a long query from the last response it saw. This one uses plain ExecutePlan, which Spark 4 servers fully support, but a flaky network on an hour-long query loses the query. It is the first thing to add.
User-defined functions. A UDF in Connect is serialized code shipped to the server: pickled Python, or a Scala closure with its JARs uploaded through AddArtifacts. It only works when the server can run the client's language, and there is no OCaml runtime on a Spark executor. This client will never have UDFs; it has the whole SQL function library and higher-order functions with lambdas. Code that already lives on the server is a different matter: the desk's market-data adapter is Python installed on the executors, and any client can call it by name.
Streaming, MLlib and pipeline commands. The messages are generated and reachable, but each is a stateful API rather than a plan.
Dictionary-encoded or compressed Arrow batches. Spark does not send them by default; the decoder raises if it ever meets one.

Each item is either resilience or a place where the client has to be a runtime instead of a plan builder. The line between those halves is the line Connect draws. The first half is small enough to build in a language the Spark project will never target. The second half is where the Go, Swift and Rust clients are spending their effort, and it is the half to check before building on any of them.

Hosting an Entire Open Lakehouse on Port 8080

lnc — Wed, 14 Oct 2026 05:21:07 +0000

Most of my job, at least the part people see, is running Spark code in front of other people and promising it'll work. The catch is that I almost never demo the stable stuff. Nobody wants a talk about the version of Spark they're already running; they want the release that came out last month, or the feature that's still a JIRA ticket, so the thing I'm showing is usually newer than its documentation and occasionally newer than any release. Iceberg ships on its own schedule, both of them depend on an exact Hadoop and an exact AWS SDK, and the whole arrangement can quietly stop working between the week I build a demo and the week I give it without a single line of my code changing.

So in December 2025 I started lakehouse-at-home: somewhere real to install new versions of Spark and Iceberg and find out what broke before an audience did. This is how it got built, one service and one problem at a time. Each section adds to the architecture diagram; new pieces are highlighted, and everything built before them fades into the background.

Why on my own machine

Honestly, part of it was curiosity about the hardware. My desktop is a Ryzen 7 7800X3D with 64GB of RAM and an RTX 5070 Ti, which is a ridiculous amount of machine to spend mostly on a browser, and I wanted to know whether it could carry a real lakehouse: not single-node Spark reading a Parquet file, but the actual set of services a data platform runs on, all at once, talking to each other the way they would in production.

The rest was control. Managed platforms give you the versions they support, not the one that shipped last week, and certainly not a build from Spark's master branch, which is where I ended up more than once. A demo also has to behave the same way every single time; while preparing a recording in March I wrote the requirement down more bluntly than any design doc would: "keep in mind with demos theres a certain amount of cleanup and restarting needed for each session and predictable behaviour". Tearing everything down and bringing it back identical is trivial when the whole lakehouse lives on a box I own, and it costs nothing, which matters when you do it several times a day.

And I had a longer game in mind. If I was going to run a lakehouse at home anyway, it could eventually hold data that actually lives at home: readings from my IoT devices, my own documents, the kind of personal data I'd rather not ship to someone else's cloud to query. Local wasn't only the cheap option; it was the only one where the data could eventually be mine.

Starting from the data

I started from the tables rather than the compute, because the tables were the point. I wanted Iceberg and Delta side by side, reading and writing the same kinds of data, so that when a new release changed how one of them behaved I'd see the difference rather than read about it. (That paid off. Delta 4.0.1, for example, turned out not to load on Spark 4.1 at all, which I only found out because I tried.)

An open table format is really three things pretending to be one: an engine that reads and writes, a catalog that records which tables exist and which snapshot of each is current, and storage for the data files themselves. Each needs somewhere to live.

The simplest catalog that behaves like a real one is Iceberg's JDBC catalog, which keeps table metadata in an ordinary relational database, so I used Postgres, running natively on the host. For storage I wanted S3 semantics rather than a local directory, because every real lakehouse sits on object storage, so I picked SeaweedFS, which speaks the S3 API, and ran it natively too. On the Spark side, wiring that up is a handful of lines:

spark.sql.extensions              org.apache.iceberg.spark.extensions.IcebergSparkSessionExtensions
spark.sql.catalog.iceberg         org.apache.iceberg.spark.SparkCatalog
spark.sql.catalog.iceberg.type    jdbc
spark.sql.catalog.iceberg.uri     jdbc:postgresql://localhost:5432/iceberg_catalog
spark.sql.catalog.iceberg.warehouse  s3a://lakehouse/warehouse

And that's where it got hard, because Spark doesn't talk to S3 by itself. It goes through Hadoop's s3a connector, which is compiled against one specific version of the AWS SDK, and Iceberg's runtime has to agree with both. Get any of them slightly wrong and nothing tells you which one; you get a ClassNotFoundException, or a NoSuchMethodError from somewhere deep in the SDK, or a credentials failure that looks exactly like a credentials failure and isn't. Getting Spark to authenticate against SeaweedFS at all took a lot of trial and error with the s3a settings:

spark.hadoop.fs.s3a.endpoint                http://localhost:8333
spark.hadoop.fs.s3a.path.style.access       true    # SeaweedFS isn't AWS: no virtual-host bucket URLs
spark.hadoop.fs.s3a.connection.ssl.enabled  false   # plain HTTP on localhost
spark.hadoop.fs.s3a.impl                    org.apache.hadoop.fs.s3a.S3AFileSystem

and then a lot more trial and error with the JARs, until the combination that worked was pinned so hard that the first README warns about it in the troubleshooting section, parenthetical and all: Hadoop 3.4.1 → AWS SDK v2.24.6 (not newer!). The download script from that week still pulls both AWS SDK bundles, the v1 bundle and the v2 one, because by that point I wasn't going to argue with anything that worked. All told it came to "~860MB of dependencies (exact versions required for compatibility)":

Component	Pinned to	Why
Spark	4.0.1 (Scala 2.13, Java 17)	the thing under test
Iceberg runtime	`iceberg-spark-runtime-4.0_2.13-1.10.0`	must match Spark's minor version and Scala version
Hadoop	`hadoop-aws-3.4.1`	provides `s3a`
AWS SDK v2	`bundle-2.24.6`	the exact version Hadoop 3.4.1 was built against
AWS SDK v1	`aws-java-sdk-bundle-1.12.780`	belt and braces
Postgres JDBC	`42.7.4`	for the Iceberg catalog

Most of what goes wrong when you upgrade a lakehouse isn't the engine; it's the dependency graph around it, and the only way to learn that graph is to break it somewhere safe. The project's agent guide still has a section called Critical Versions: Do not change without testing, which is the same lesson written for machines.

Once it worked, it was an ordinary lakehouse, and the first thing any test of it does is the ordinary thing:

CREATE NAMESPACE IF NOT EXISTS iceberg.bronze;

CREATE TABLE iceberg.bronze.orders (
    order_id    STRING,
    customer_id STRING,
    amount      DOUBLE,
    created_at  TIMESTAMP
) USING iceberg;

The last decision in that first week was where everything should run, and it shaped the next nine months more than anything else. Spark went in Docker, because I'd want to swap its versions constantly. Postgres and SeaweedFS stayed native on the host, because they were storage, and storage should outlive the experiments. The first compose file still shows the argument I had with myself about how the two sides would reach each other: a bridge network, published ports and a host.docker.internal gateway, all commented out, and above them the line that won:

spark-master:
  image: apache/spark:4.0.1-scala2.13-java17-python3-r-ubuntu
  network_mode: "host"
#    ports:
#      - "8090:8080"  # Spark Master Web UI
#      - "7077:7077"  # Spark Master
#    networks:
#      - lakehouse
#    extra_hosts:
#      - "host.docker.internal:host-gateway"

Host networking made everything simple on my machine, and it's also the reason the title of this post is a port number. Every container shared the host's ports with every native service, so from then on each new component had to find a port nobody else was using. Keep an eye on the port numbers in the diagrams; they're the plot.

A catalog I had to build

Once I had tables, I wanted the kind of catalog people actually run in production: one that governs access, vends credentials, and lets more than one engine find the same tables. That meant Unity Catalog OSS, running alongside the Postgres catalog rather than replacing it, so both paths stayed testable.

The problem was that the features I wanted were ahead of the published images. When I was preparing an episode of OverArchitected in March, the release I needed didn't have a Docker image yet, so I cloned the repository, built the image myself, and ran a 0.5.0 snapshot from source. It was the first time the stack ran something that hadn't been released, and it wasn't the last. It also needed a port: its default was 8080, which Spark's UI already had, so it moved to 8081. From Spark's side it's just a second catalog, talking Iceberg's REST protocol:

spark.sql.catalog.unity               org.apache.iceberg.spark.SparkCatalog
spark.sql.catalog.unity.catalog-impl  org.apache.iceberg.rest.RESTCatalog
spark.sql.catalog.unity.uri           http://localhost:8081/api/2.1/unity-catalog/iceberg

Streams: Kafka and Zookeeper

Batch tables only tell half the story, so the next piece was streaming: Kafka for the events, with Zookeeper to coordinate it.

This part was harder than it had any right to be. When I finally sat down with it in January, the prompt was the whole mood of the week: "debug my kafka setup, why can't spark read it?" The answer was about eight small things at once. The streaming job never ran from_json() on Kafka's value column, so it was trying to aggregate raw bytes; a column was referenced as total.revenue instead of total_revenue; one alias was the wrong case; a debug query was blocking the stream. Kafka itself wasn't on host networking, so Spark couldn't reach it, and once it was, its connection string to Zookeeper was wrong. And Zookeeper's admin server, which listens on 8080 by default, was fighting Spark's web UI for the same port. Zookeeper lost, and moved to 9999.

The fix for the first of those is the part every Kafka tutorial skips: Kafka hands Spark bytes, and you have to tell it what they mean.

kafka_df = (
    spark.readStream
    .format("kafka")
    .option("kafka.bootstrap.servers", "localhost:9092")
    .option("subscribe", "orders")
    .option("startingOffsets", "latest")
    .load()
)

parsed = kafka_df.select(
    f.from_json(f.col("value").cast("string"), event_schema).alias("event"),
    f.col("timestamp").alias("kafka_timestamp"),
)

Weeks later I was still citing Kafka as the cautionary tale; when I worried I was adding things too fast, the reason I gave was "how much trouble even setting up and testing kafka was".

Data that misbehaves

A lakehouse with no data in it proves nothing, and a lakehouse full of clean data proves less than you'd think: every pipeline looks correct when nothing is ever late, duplicated or missing. I needed data that behaved like the real thing, which is to say badly, and at a volume that would actually exercise the stack.

Rather than invent a domain, I borrowed one. Databricks' Casper's Kitchens simulates a ghost-kitchen delivery business, and I asked the agent to study how its generators worked and rebuild the idea for this stack ("please consult me first", I added, because I wanted to understand the shape of the data before it existed). What came out is a small Python package that simulates the full life of a food order:

Each step takes a Gaussian amount of time, demand follows the day, and some brands are growing while others decline, so there are real trends for a pipeline to find:

@dataclass
class ServiceTimes:
    """Gaussian distribution parameters for order lifecycle phases (in minutes)."""
    order_to_kitchen_start: tuple[float, float] = (2.0, 0.5)
    kitchen_prep: tuple[float, float] = (15.0, 5.0)
    kitchen_to_driver: tuple[float, float] = (5.0, 2.0)
    pickup_to_delivery: tuple[float, float] = (20.0, 8.0)
    driver_ping_interval_sec: int = 60

Demand peaks at noon and again from six to eight in the evening, and Fridays and Saturdays run 25 to 35 percent busier than a weekday. And then, on purpose, it breaks things:

@dataclass
class ChaosConfig:
    """Data quality issue injection settings."""
    enabled: bool = True
    null_rate: float = 0.05            # 5% of fields nullified
    late_event_rate: float = 0.03      # 3% out-of-order events
    duplicate_rate: float = 0.02       # 2% duplicate events
    malformed_json_rate: float = 0.01  # 1% malformed body

Chaos	What it tests
Null fields	schema enforcement, defaults, data quality checks
Late (out-of-order) events	watermarks, event time vs processing time
Duplicate events	idempotency, deduplication
Malformed JSON	parsing, dead-letter handling

The same generator feeds both halves of the stack:

./lakehouse testdata generate --days 7     # history as Parquet
./lakehouse testdata load                  # into Iceberg bronze tables
./lakehouse testdata stream --speed 60     # replay into Kafka, an hour per minute
./lakehouse testdata stats

The duplicates and late events are there because the interesting failures only show up when the data misbehaves, and months later, when I tested AUTO CDC, cancellations and duplicate creation events were exactly the cases the feature existed for.

Spark, twice

The point of all of this was testing Spark versions, and that's the part I designed most deliberately. Instead of upgrading 4.0 to 4.1 in place, the stack runs them side by side: each version gets its own compose file, its own container names, its own image (with the Java version the official image ships), and its own block of ports, so both clusters can be up at once and the same job can be submitted to either.

	Spark 4.0.1	Spark 4.1.0
Compose file	`docker-compose.yml`	`docker-compose-spark41.yml`
Image	`apache/spark:4.0.1-scala2.13-java17-…`	`apache/spark:4.1.0-scala2.13-java21-…`
Master	7077	7078
Master UI	8080	8082

./lakehouse start spark                  # Spark 4.0.1
./lakehouse start spark --version 4.1    # Spark 4.1.0, alongside it

That turned out to be the organising idea for a lot of what followed. Spark Declarative Pipelines arrived in 4.1, so the workshop I built on the stack became a pair of notebooks: the same medallion pipeline written imperatively on 4.0 and declaratively on 4.1, running against the same Iceberg tables at the same time.

Orchestration: Airflow

Pipelines that only run when I type spark-submit aren't really pipelines, so in January the stack got Apache Airflow. The first version was Airflow 2.8, and within a week I upgraded it to 3.1, which was educational in the way major versions are: the webserver became an api-server, schedule_interval became schedule, BashOperator moved into a providers package, and the health endpoint moved.

In the middle of that upgrade, the Java version in one image flipped from 17 to 21 and back again, and my reaction at the time was correct, if not patient: "why do we keep switching java versions? this is a fundamental change we can't keep expecting users to keep up with". The rule that came out of it is that Java versions are whatever the official images ship, and nobody changes them.

The more useful correction was conceptual. The first architecture diagrams drew Airflow talking to Iceberg, and I asked the question that should have been asked earlier: "does airflow being linked to the iceberg api really make sense in the architecture diagrams? what is airflow really orchestrating in ths stack? just spark pipelines right?" It was. Airflow doesn't touch the data; it tells Spark when to, and the diagram got simpler once it said so:

The stack ships three DAGs:

DAG	What it does
`lakehouse_medallion_pipeline`	checks Kafka is reachable, branches on which Spark version to use, runs the medallion job on 4.0 or 4.1, then verifies the tables
`sdp_pipeline`	preflight, runs a declarative pipeline from its `spark-pipeline.yml` spec, verifies, then maintains
`iceberg_maintenance`	the chores nobody remembers: `rewrite_data_files` to compact small files, `expire_snapshots`, `remove_orphan_files`

The branch in the first DAG is my favourite detail in the repo, because the version-testing idea reaches all the way into orchestration:

choose_spark = BranchPythonOperator(
    task_id="choose_spark_version",
    python_callable=choose_spark_version,
)

run_pipeline_spark41 = BashOperator(
    task_id="run_pipeline_spark41",
    bash_command="""
        docker exec spark-master-41 /opt/spark/bin/spark-submit \
            /scripts/pipelines/pipeline_spark41.py
    """,
)

check_kafka >> choose_spark >> [run_pipeline_spark41, run_pipeline_spark40]
[run_pipeline_spark41, run_pipeline_spark40] >> verify_tables

Running declarative pipelines from Airflow was clumsier at first, because there was no operator for them; that changed in March, and I'll come back to it once SDP itself has been introduced.

One CLI, built for agents

By now there were two Spark clusters, a streaming stack, two catalogs, an orchestrator and a pile of native services, and remembering which compose file, which ports and which container names went with which version had become a job in itself. So everything went behind a single entry point, ./lakehouse, a bash CLI that wraps Docker Compose and hides the bookkeeping:

./lakehouse setup                        # prerequisites, Postgres catalog, JARs
./lakehouse preflight                    # do .env, Spark config and the catalog agree?
./lakehouse start all --version 4.1      # Spark + Kafka, then wait until they answer
./lakehouse start airflow                # optional services, one at a time
./lakehouse status --json                # machine-readable, for scripts and agents
./lakehouse logs spark-master
./lakehouse stop all

Run it with no arguments and it explains itself, including one line that would matter later:

$ ./lakehouse help
Lakehouse Management CLI

Usage: lakehouse <command> [options] [--version 4.0|4.1]

Commands:
  setup               Validate environment and install dependencies
  check-config        Validate credentials between .env and spark-defaults.conf
  preflight           Run preflight checks before starting services
  migrate [--dry-run] Run database schema migrations
  status              Show all services status
  start [service]     Start service (spark|kafka|airflow|unity-catalog|all)
  ...
  testdata <cmd>      Test data generation (generate|stream|load|stats|clean)
  browsedata [table]  Browse Iceberg table rows (default: iceberg.bronze.orders)

Services:
  ...
  all            Spark + Kafka (not Unity Catalog or Airflow)

all means the core services, which I discovered the hard way on the morning of a recording, after a reboot, when ./lakehouse start all brought back Spark and Kafka and not the catalog or MLflow I was about to demo.

The deeper reason for one CLI was that I wanted the stack to be agent-first. Early on I wrote down the goal: "I want people to feed an agent the repo, and be able to get up and running locally or on a cloud provider like AWS with minimal hitches." At the time, coding agents were far more reliable driving a well-behaved command line than reconstructing a sequence of docker compose invocations from a README, so the CLI was designed for them as much as for people. Commands fail with real exit codes and say what to do next, and status --json exists so an agent can read the state of the stack instead of scraping coloured terminal output.

Some features came straight from me asking the obvious question, like "is there a cli test process in setup that checks if ports are available for the services?"; there wasn't, so now start checks every port it's about to need and refuses to run if something else is holding one:

check_port_available() {
    local port=$1
    local expected_container=$2

    if ! nc -z localhost "$port" 2>/dev/null; then
        return 0                      # port is free
    fi
    # port is in use: fine only if it's the container we expect
    if [ -n "$expected_container" ] && \
       docker ps --format '{{.Names}}' | grep -q "^${expected_container}$"; then
        return 0
    fi
    return 1                          # blocked by something else
}

The documentation is split the same way. CLAUDE.md works as an agent's guide to the project. AGENTS.md describes itself as a "compressed reference for AI agents. Always in context - no skill invocation required": Spark 4.1's breaking changes, standard imports, API quick references, the things an agent gets wrong when it's working from older training data. Deeper, task-specific knowledge went into skills, which grew into their own repo, lakehouse-skills. That started because the agent kept stumbling over SDP's API, and I asked it, fairly pointedly, to write down what it had learned ("you had some struggles using the official API. can you create an SDP.md skills file"). Then I tested the skill the only way that means anything: a fresh agent session authored pipelines against the stack using nothing but that file, and wherever its output was wrong, the skill got fixed.

There's one design choice in the CLI I'd make differently now. Its definition of a healthy service is the one almost every homelab script settles on:

is_service_healthy() {
    local port=$1
    nc -z localhost "$port" 2>/dev/null
}

A service is up if something answers on its port. Here's what status says on my machine as I write this, with only some of the stack running:

$ ./lakehouse status
================================
LAKEHOUSE STATUS
================================

System Services:
✓ PostgreSQL (port 5432)
✓ SeaweedFS (port 8333)

Unity Catalog (optional):
✓ unity-catalog
✓ Unity Catalog REST API (port 8081)

Spark 4.1 Containers:
✗ spark-master-41
✗ spark-worker-41

Airflow Containers:
✓ airflow-webserver
✗ Airflow UI (port 8085)
✓ airflow-scheduler

Ports:
  Unity Catalog:    8081 (REST API + Iceberg endpoint)
  Spark 4.0:        7077, 8080 (UI), 8081 (Worker UI)
  Spark 4.1:        7078, 8082 (UI), 8083 (Worker UI)

Two things in there are the whole post in miniature. The port table lists 8081 twice: Unity Catalog moved there to get out of the Spark UI's way, and Spark 4.0's worker UI has been sitting on 8081 by default the entire time. And Airflow's container is up while its UI isn't answering, which the human-readable status reports honestly and the JSON version doesn't:

"airflow": {
  "webserver": true,
  "scheduler": true
},
"all_healthy": false

A port check tells you something is listening, not that it's the thing you think it is, at the version you think it is, and a container check tells you something is running, not that it's serving. On a machine where every service shares one set of ports, that distinction eventually matters. It mattered in June, when a Spark "bug" I'd had written up with a minimal reproduction turned out to be an old Spark Connect server that had been answering on the port the whole time. That one gets its own post.

Credentials, everywhere

A lakehouse at home has fewer secrets than a real one, but it has the same shape of secrets, and they multiply faster than you'd expect. The Iceberg catalog needs a Postgres username and password. Spark needs S3 keys for SeaweedFS. Unity Catalog needs its own copy of the storage credentials so it can hand them out. Airflow has an admin login, MLflow has a database, and every one of those values lives in more than one place.

The pattern the repo settled on is boring on purpose. Real values live in files that are never committed (.env, spark-defaults.conf, the Unity Catalog server.properties), each with a checked-in .example template full of obvious placeholders like your_access_key, and .gitignore opens with a section literally titled Credentials and Secrets. The part that took longer to learn was that the same secret stored in two files will eventually disagree, and the error you get when it does never says so; Spark reports an S3 failure, or Iceberg reports a missing catalog. So the CLI grew a command that checks the copies against each other, and against the placeholders nobody replaced:

$ ./lakehouse check-config
================================
CONFIGURATION CHECK
================================

1. Checking .env file...
   ✓ .env configuration valid

2. Checking spark-defaults.conf...
   ✓ Credentials consistent

3. Testing database connection...
   ✓ PostgreSQL connection successful

4. Testing S3 endpoint...
   ✓ S3 endpoint responding

It's also where I found my own worst habit in the codebase. An early version of the CLI connected to the catalog as a hardcoded Postgres user, and the user was me: my own username, baked into a script I was about to tell strangers to run. Reading credentials from .env instead of hardcoding them was one of the first fixes anyone would need to run the stack at all, and detect-secrets now sits in the pre-commit chain so the next version of that mistake doesn't make it into a commit.

The catalogs raise the stakes, because a catalog's job is partly to hold credentials so that engines don't have to. Unity Catalog can vend short-lived storage credentials to whoever asks for a table, which is the whole point of running it, but it needs a complete set of storage credentials to vend; in one setup it quietly skipped configuring a bucket until every S3 field was filled in, including a placeholder session token that a local SeaweedFS has no use for. And the classic Spark layout has the opposite problem: because every node that runs a driver needs the catalog, the catalog password and the S3 keys end up mounted into the master and every worker. That, more than anything, is why Spark Connect ended up in the stack.

Testing it, and CI

A stack whose whole purpose is catching breakage needs to catch its own. The repo's tests come in two layers: fast unit tests (project structure, DAG definitions, pipeline transformations, the CLI's security guarantees) and integration tests that run against real services, one file per component: Iceberg CRUD, SeaweedFS, Kafka streaming, Unity Catalog, Airflow, the declarative pipelines across file and table formats, and a test_fresh_install that checks the thing I kept promising people, that it works from a clean clone. The integration tests read like a lakehouse tutorial, which is roughly the point:

@pytest.mark.integration
class TestIcebergCRUD:
    def test_create_table(self, spark_with_postgres):
        spark = spark_with_postgres
        spark.sql("CREATE NAMESPACE IF NOT EXISTS iceberg.test_crud")
        spark.sql("""
            CREATE TABLE IF NOT EXISTS iceberg.test_crud.orders (
                order_id STRING, customer_id STRING,
                amount DOUBLE, created_at TIMESTAMP
            ) USING iceberg
        """)
        tables = spark.sql("SHOW TABLES IN iceberg.test_crud").collect()
        assert "orders" in [row.tableName for row in tables]

CI runs on GitHub Actions, as three workflows that hand off to each other (with a copy of the CI workflow on a self-hosted Gitea, so the same checks can run on my own hardware):

Inside ci.yml, the jobs run in stages that each gate the next:

The last stage is the version-testing idea again, this time for the stack itself: the same checks run against both Spark images.

spark-matrix:
  name: Spark ${{ matrix.spark-version }} Tests
  strategy:
    matrix:
      spark-version: ['4.0', '4.1']
      include:
        - spark-version: '4.0'
          image: apache/spark:4.0.1-scala2.13-java17-python3-r-ubuntu
        - spark-version: '4.1'
          image: apache/spark:4.1.0-scala2.13-java21-python3-r-ubuntu
          compose-file: docker-compose-spark41.yml

Around the tests sits the less glamorous machinery that makes a repo safe to hand to strangers (and their agents). Work flows from feature branches into develop, where CI runs on every pull request, and only then into master; releases are cut by release-please from conventional commit messages. The CI workflow pins its actions to commit SHAs rather than tags (going back through it for this post, I noticed two exceptions I should fix: the Codecov upload in CI, and the release workflows, which still use tags like @v4), the workflow's default permissions are read-only, and a pre-commit hook chain runs detect-secrets, bandit, shellcheck, hadolint, ruff and black before anything reaches a branch. That hardening came out of the same January week where I admitted I was scared I was going too fast; the test suite's own test_security.py still guards the fixes it produced, like removing an eval from the CLI and closing a SQL-injection path in lakehouse migrate.

CI also taught me one thing about the outside world. The Airflow image downloads a Spark release tarball, and once Spark 4.1.1 superseded 4.1.0, the old tarball moved from Apache's fast mirrors to the slow archive server, and the build started timing out. Nothing in my repo had changed; the internet had.

Declarative pipelines and real time

The last layer is the reason I needed all the others: the new Spark features I actually demo.

Spark Declarative Pipelines arrived in 4.1, which is exactly why two Sparks mattered. Instead of writing the order of operations yourself, you declare the tables you want and how each is derived, and SDP works out the graph. In this stack a small spec points at a Python file:

name: lakehouse_pipeline
libraries:
  - glob:
      include: pipeline_sdp.py
catalog: iceberg
database: bronze

and the Python file is a set of decorated functions, from a streaming bronze table fed by Kafka up to gold aggregates:

from pyspark import pipelines as dp
from pyspark.sql import functions as f

@dp.table(name="bronze.orders_streaming")
def orders_streaming():
    ...  # the Kafka readStream + from_json from earlier

@dp.materialized_view(name="gold.hourly_metrics")
def hourly_metrics():
    orders = spark.table("iceberg.silver.orders_enriched")
    return (
        orders.filter(f.col("event_type") == "order_created")
        .groupBy("event_date", "event_hour", "location_id", "city_name")
        .agg(
            f.count("order_id").alias("order_count"),
            f.sum("order_total").alias("total_revenue"),
            f.avg("order_total").alias("avg_order_value"),
        )
    )

It also turned into a benchmark: on the stack's own Iceberg, Postgres, SeaweedFS and Kafka, a million-row pipeline took 11.73s through spark-pipelines against 7.75s as a raw spark-submit, a fixed four seconds or so of orchestration that's noticeable on small jobs and irrelevant on large ones. SDP is a pipeline abstraction, not a performance tool, and I'd rather know that than guess at it on stage.

Scheduling SDP: `SparkPipelinesOperator`

Running a pipeline by hand is a demo; running it every night is a platform, which brings Airflow back in. My first SDP DAG did what everyone did in early 2026: it wrapped the spark-pipelines CLI in a SparkSubmitOperator, pretending the CLI was a Spark application:

# Before: SDP smuggled through spark-submit
SparkSubmitOperator(
    task_id="run_pipeline",
    application="/usr/local/bin/spark-pipelines",
    application_args=["run", "--spec", "/scripts/pipelines/spark-pipeline.yml"],
    conn_id="spark_default",
)

It mostly worked, and when it didn't, it failed in confusing ways, because SDP pipelines don't really run through spark-submit. The spark-pipelines CLI dispatches through Spark Connect when SPARK_REMOTE is set and through spark-submit otherwise, and in Spark 4.1 its shell wrapper took the JVM path even in Connect mode: it tried to start a second Spark Connect server on port 15002, next to the one already there, and rejected the --master and --deploy-mode flags the operator was passing. (Yes, 15002 again.)

In late March a dedicated SparkPipelinesOperator landed in Airflow's Spark provider, and it's genuinely much easier. You hand it the pipeline spec and a command, and it works out the rest:

# After: an operator that knows what a pipeline is
SparkPipelinesOperator(
    task_id="run_pipeline",
    pipeline_spec="/scripts/pipelines/spark-pipeline.yml",
    pipeline_command="run",          # or "dry-run" to validate the graph first
    conn_id="spark_default",
)

Under the hood it skips the shell wrapper and calls SDP's Python entry point directly, so it takes the right path in both modes. On top of that it adds the things you'd want from any Airflow operator. The spec, command, Spark conf and environment variables are all templated, so one DAG can run a pipeline per date partition. pipeline_command is validated, so a typo fails at parse time instead of at 2 a.m. And its logs say which Spark it actually talked to, which, given everything else in this post, I appreciate more than I can say.

That also made patterns cheap that used to be fiddly. The stack's demo DAGs include a dry-run → run chain, so a broken dependency graph is caught before any cluster time is spent:

validate = SparkPipelinesOperator(task_id="validate", pipeline_spec=SPEC, pipeline_command="dry-run")
run      = SparkPipelinesOperator(task_id="run",      pipeline_spec=SPEC, pipeline_command="run")
validate >> run

alongside multi-pipeline chains, sensor-triggered runs that wait for data to land in Kafka, and per-pipeline resource tuning. The repo vendors a copy of the operator so its DAGs work regardless of which provider version is installed, with an import that prefers the upstream one when it's there.

Converging streaming and batch

The usual answer to "we need real time" is a second engine. Spark does the batch work, Flink does the low-latency streaming, and now there are two runtimes, two codebases, two sets of semantics for the same business logic, and two things to keep upgraded. For a stack on one desktop, whose whole purpose is testing Spark, I wanted exactly one engine, and I wanted batch and streaming to be the same code wherever possible.

Spark gets surprisingly close. In the declarative pipeline above, the Kafka-fed bronze table is a streaming table and the silver and gold layers are materialized views, in the same file, over the same Iceberg catalog, run by the same spark-pipelines run. Outside of SDP, the same DataFrame logic can sit anywhere on the latency spectrum by changing one line, the trigger:

query = (
    parsed_orders.writeStream
    .format("kafka")
    .option("topic", "orders_out")
    .outputMode("update")
    .trigger(availableNow=True)           # incremental batch: process what's there, then stop
  # .trigger(processingTime="1 second")   # micro-batch: latency measured in seconds
  # Real-Time Mode:                       # continuous: latency measured in milliseconds
)

Real-Time Mode is what makes the one-engine argument hold at the low-latency end. Instead of scheduling a new micro-batch every interval, it keeps long-running tasks processing records as they arrive and only checkpoints periodically. With the stack's Kafka producing 500 events a second across 50 keys, a stateless Kafka-to-Kafka pipeline on Spark 4.1, on this desktop, measured a median latency of 17 milliseconds, steady across six runs:

Percentile	Latency (6 runs)
p50	16 to 18 ms
p95	20 to 24 ms
p99	21 to 26 ms

That's the latency range people usually reach for a second engine to get. But in 4.1 there were real limits, and the stack found all of them:

4.1 limitation	What it meant
Stateless only	the RTM operator allowlist permits projections, filters and broadcast joins, but not aggregations, shuffles or state stores; no running counts, no deduplication
No Python UDFs	Python UDF execution isn't on the allowlist
No PySpark keyword	`trigger(realTime=...)` didn't exist in PySpark 4.1
Console sink doesn't flush	use a Kafka sink to see output
`current_timestamp()` is per checkpoint, not per row	latency has to be measured outside Spark, by a consumer comparing wall clocks
Every stage runs at once	a 16-partition Kafka source needs 16 task slots just to read; stateful plans need far more

The PySpark gap was the most awkward for me, since everything else in the stack is Python. The quickstart got around it by reaching through Py4J into the JVM and setting Scala's trigger directly:

def with_real_time_trigger(dsw, checkpoint_interval="5 minutes"):
    """Apply RealTimeTrigger to a DataStreamWriter via the JVM gateway."""
    jvm = dsw._spark._jvm
    jTrigger = jvm.org.apache.spark.sql.execution.streaming.RealTimeTrigger.apply(
        checkpoint_interval
    )
    dsw._jwrite = dsw._jwrite.trigger(jTrigger)
    return dsw

which works on a classic Spark session and is impossible over Spark Connect, where the client has no JVM to reach into; one more reason the Connect migration needed care.

Spark 4.2 added the native PySpark trigger, and 4.3 adds stateful operators to Real-Time Mode. That upgrade gets a post of its own.

Building Iceberg for a Spark that doesn't exist

In May, the contributor who wrote AUTO CDC (SDP's new change-data-capture flow) asked me to put it through its paces before it went public. The feature had landed on Spark's master branch, and that meant it existed in exactly one place. master is where development for Spark's next major version happens, so it builds as 5.0.0-SNAPSHOT: not a release, not even a preview, just whatever is on the branch that day. Before any Apache Spark release, the project cuts a release candidate (an "RC"), a build the community downloads, tests and votes on; only a passing vote turns it into a release. AUTO CDC hadn't made it onto a release branch yet, so there wasn't even a release candidate to try. The stack grew a third Spark, and it had to be built from source first:

cd ~/spark-src
./dev/make-distribution.sh --pip --tgz --connect -Phive -Phive-thriftserver   # 30 to 45 minutes
./build/mvn install -DskipTests -Phive -Phive-thriftserver                    # publish to ~/.m2

That was the easy half. AUTO CDC writes by merging changes into its target row by row, so the target needs a table format that supports row-level operations; point it at plain Parquet and it refuses with AUTOCDC_TARGET_DOES_NOT_SUPPORT_MERGE. I needed Iceberg. But Iceberg's Spark integration is built separately for each Spark minor version, and the newest one in Iceberg's repository was spark/v4.1. The released 4.1 runtime didn't just lack support for Spark 5.0; it failed to load at all, because Spark 5.0 had deleted an interface Iceberg compiles against. There was nothing to download, so I built one.

The port started by pointing Iceberg's build at the snapshot instead of the last release:

 # gradle/libs.versions.toml
 spark40 = "4.0.2"
-spark41 = "4.1.1"
+spark41 = "5.0.0-SNAPSHOT"

and then compiling, reading the errors, and fixing what Spark had changed underneath it. Most of the work fell into three kinds of API drift:

What changed on `master` (the 5.0 snapshot)	What broke in Iceberg	What the port did
The DSv2 `View` connector interface was removed (replaced by `ViewInfo` and `TableViewCatalog`)	Iceberg's view support, its catalogs' view methods, and its SQL extensions, which are built on views	removed view support entirely: a tables-only port, which is most of the 534 deleted lines
New abstract methods: `Reducer.resultType()`, and nanosecond timestamp getters on `SpecializedGetters`	Iceberg's partition functions and row readers no longer compiled	implemented them (integer result types; nanosecond timestamps throw)
`StagedTable` and `TruncatableTable` both gained a default `reportDriverMetrics()`	Java refuses a class that inherits two unrelated defaults	overrode it explicitly in Iceberg's staged tables

The last one is my favourite, because nothing was wrong with either interface on its own; Iceberg's staged tables implement both, and the moment both grew a default method with the same name, Java couldn't choose between them:

@Override
public CustomTaskMetric[] reportDriverMetrics() {
  return new CustomTaskMetric[0];
}

Then the storage layer did to me exactly what it had done in December. Spark 5.0 moved to Hadoop 3.5.0, so the carefully pinned s3a stack from the first week was wrong again, and the new combination needed hadoop-aws-3.5.0, AWS SDK bundle 2.35.4, and a dependency that hadn't existed before, analyticsaccelerator-s3. Even Kafka needed its JARs copied into the distribution's own jars/ directory before the new server would register it as a data source. Every new Spark reopens the dependency graph from the first week, and the only question is whether you find out on your own machine or on stage.

The runtime I ended up with kept the name iceberg-spark-runtime-4.1_2.13 (the module wiring was reused), which is the most misleading filename I own. It can't do Iceberg views or SQL procedures. It could do exactly one thing well, native row-level MERGE into Iceberg tables, which is the one thing AUTO CDC needed, and with it I ran the feature against Kafka streams, Debezium change feeds and the ghost-kitchen orders, cancellations and all, weeks before it shipped. (I also asked whether I should port Delta the same way. The answer was no: AUTO CDC needs a table that implements Spark's row-level operations interface, and Delta does its merges through its own command instead, so a ported Delta still couldn't have been a target.) The findings are in AUTO CDC in Spark Declarative Pipelines.

The coda is that this wasn't a one-off. When Spark 4.2 shipped for real in July, Maven Central's Iceberg runtimes still stopped at 4.1, and I built Iceberg locally again to run the stack on it. Testing a new Spark release, it turns out, very often means building a little of the ecosystem that hasn't caught up to it yet.

Why run Spark Connect at home at all

Spark Connect splits Spark in two: a thin client that builds query plans and sends them over gRPC, and a server that owns the driver and actually runs them. On a single desktop that sounds like complexity for its own sake, since the client and the server are the same machine. I ran it anyway, because the classic layout had three problems that got worse the bigger the stack got:

Code had to live beside the driver. A pipeline couldn't be submitted from my laptop or from CI; it had to be copied into the Spark master container first, which is why the CLI grew copy, run and delete wrappers around spark-submit.
Credentials lived everywhere a driver could. The catalog password and S3 keys were mounted into the master and every worker.
The application and the platform shared one environment. Upgrading Spark changed the Python environment and JARs every job ran in, which is exactly backwards for a stack whose purpose is upgrading Spark.

from pyspark.sql import SparkSession

spark = SparkSession.builder.remote("sc://localhost:15002").getOrCreate()
spark.table("iceberg.gold.hourly_metrics").show()

There was also a practical reason: SDP leans on Connect's machinery even when you never type sc:// yourself, so Spark's newest pipeline features were going to pull me toward it regardless.

Running it locally was not free. Things that had quietly worked stopped working, and the failures were instructive:

Problem	What happened	What fixed it
Driver-only APIs	`spark.sparkContext.setLogLevel("WARN")` fails with `JVM_ATTRIBUTE_NOT_SUPPORTED`; the client has no `sparkContext`, `_jvm` or `_jsc`. An audit of one pipeline found 21 blocking calls, every one of them `setLogLevel`	log levels belong to the server; set them there
Config the client can't set	`spark.jars` and `spark.sql.extensions` are fixed when the server starts, so a client can't add Iceberg's extensions (and without them there's no `MERGE INTO` or `CALL iceberg.system.*`)	put the JARs and extensions on the server; check capabilities at client startup
A scary, harmless warning	the server logs `ClassNotFoundException: …IcebergSparkSessionExtensions` at startup, for its bootstrap session, before the JARs are loaded	verify by using `MERGE INTO` and `CALL`, not by reading the log
Connectors not registering	Kafka's JARs on `extraClassPath` weren't enough for a standalone Connect server to find the data source	copy them into the server's own `jars/` directory
Checkpoints	streaming checkpoints over S3A against SeaweedFS failed	keep checkpoints on `file://` paths on the server
One port, many servers	Connect's default port is 15002, and more than one Spark on the same machine wants it; Spark 4.1's shell wrapper could even start a second Connect server that fought the first	one server per port, explicitly; and check which server answered

When I finally moved the stack's own medallion pipeline to Connect, on Spark 4.2 with a locally built Iceberg runtime, every one of its ten Iceberg tables came out identical to the classic run, down to the content hashes (1,027,129 rows in bronze.orders alone). The last row of that table is the one that bit hardest, and it's the story of the next post.

Experimental agents

The OverArchitected episode had a premise I took more seriously than it deserved: we're lazy; can AI manage the lakehouse for us? So the stack grew three experimental agents, each a small service built on MLflow, with every model call routed through MLflow's AI Gateway and every step recorded by MLflow Tracing.

Agent	Job	Tools it gets
Pipeline Guardian	the compaction agent: inspects Iceberg table health and decides when maintenance is due	`inspect_table` (row counts, snapshots, file counts), `check_quality` (null rates, freshness, duplicates), `run_maintenance`, `query_table`
Analyst	answers business questions over the gold layer ("what's the delivery performance in San Francisco this week?")	natural language to Spark SQL, then back to prose
Autopilot	watches the streaming pipeline and tries to fix it: throughput, latency, null spikes, schema drift, compaction when file counts climb	all of the above, in a monitoring loop

The Guardian is the one I find most interesting, because table maintenance is exactly the kind of chore people forget. Iceberg tables fed by streams accumulate small files and old snapshots until reads slow down, and deciding when to compact is a judgement call about file counts, table size and query patterns. The agent makes the judgement; the tool it calls is deliberately boring:

def run_maintenance(table_name: str, action: str) -> dict:
    """Run Iceberg maintenance: compact, expire_snapshots, remove_orphans."""
    spark = get_spark()
    if action == "compact":
        spark.sql(f"CALL iceberg.system.rewrite_data_files(table => '{table_name}')")
    elif action == "expire_snapshots":
        cutoff = (datetime.now() - timedelta(days=7)).strftime("%Y-%m-%d %H:%M:%S")
        spark.sql(
            f"CALL iceberg.system.expire_snapshots("
            f"table => '{table_name}', "
            f"older_than => TIMESTAMP '{cutoff}', retain_last => 5)"
        )
    ...

Handing that judgement to a model, with a narrow set of tools and a full trace of what it looked at and why, is a reasonable experiment. It is also, and I want to be clear about this, an experiment: an agent with run_maintenance and an arbitrary-SQL tool can do real damage to real tables, which is why these live under scripts/demos and nowhere near anything I care about.

They were also the part of the stack where I pushed hardest on keeping everything open. The first version routed calls to a hosted model, and before the recording I asked for the opposite: "Can you hook up a local model into our lakehouse-stack instead of anthropic api? we go full oss if we can it will be a fun story. A US model is.. preferred". The gateway now points at models served locally through Ollama on the same desktop's GPU, and a hosted model is a one-line route change if you want one:

endpoints:
  - name: chat
    endpoint_type: llm/v1/chat
    model:
      provider: openai          # Ollama speaks the OpenAI API
      name: gpt-oss:20b
      config:
        openai_api_base: http://localhost:11434/v1

An open lakehouse whose agents phone a closed API for every decision isn't really an open lakehouse, and on a desktop with a decent GPU, it doesn't have to be.

Someone else's machine

In April the repo got its first outside contributor, and he immediately found everything that only worked because it was my machine. Denis Arnaud opened four pull requests in two days: the Unity Catalog image tag had vanished from Docker Hub; the CLI assumed GNU df, which macOS doesn't have; and the host networking and the /data/spark volume path (which exists because of how I partitioned my own drives) didn't translate to a Mac at all. He replaced them with named volumes and the very host.docker.internal gateway I'd commented out in December, and fixed a timestamp parse that Spark 4's ANSI mode rejects along the way:

# Before: under ANSI mode a failed parse raises instead of returning null,
# so the coalesce never reaches its second format.
f.coalesce(
    f.to_timestamp("ts", "yyyy-MM-dd'T'HH:mm:ss.SSSSSS"),
    f.to_timestamp("ts", "yyyy-MM-dd'T'HH:mm:ss")
)

# After: try_to_timestamp returns null on a mismatch.
f.coalesce(
    f.try_to_timestamp(events_df.ts, f.lit("yyyy-MM-dd'T'HH:mm:ss.SSSSSS")),
    f.try_to_timestamp(events_df.ts, f.lit("yyyy-MM-dd'T'HH:mm:ss"))
)

While I was reviewing them, the coding agent helpfully opened its own versions of the same fixes. My message to it was not diplomatic ("why are you creating all these new PRs without my consent either. its getting kind of messy"), and then, more to the point, "in the OSS community we give credit and encourage people's PRs, not.. just take them and implement it ourselves. lol". I closed its copies and merged his.

Where it is now

Put together, it looks like this:

The ./lakehouse CLI starts, checks and stops every box above, and GitHub Actions runs the test suite against Spark 4.0 and 4.1 on every pull request.

And for anyone who wants to find their way around the repo:

lakehouse-at-home/
├── lakehouse                     the CLI: setup · check-config · start/stop/status · testdata · test · migrate
├── docker-compose*.yml           one compose file per service: Spark 4.0, Spark 4.1, Kafka,
│                                   Unity Catalog, Airflow, MLflow, notebooks
├── config/                       Spark, Unity Catalog, Airflow, MLflow configs (.example only; live ones are gitignored)
├── dags/                         Airflow DAGs: medallion pipeline, SDP pipeline, Iceberg maintenance
├── scripts/
│   ├── pipelines/                the medallion pipeline, imperative (4.0) and declarative (4.1)
│   ├── testdata/                 the ghost-kitchen generator, with chaos
│   ├── demos/                    showcase demos and the experimental MLflow agents
│   ├── quickstarts/              Iceberg, Kafka streaming, and friends
│   ├── connectivity/             smoke tests for each service
│   └── tools/                    Kafka producer, helpers
├── notebooks/                    the 4.0 vs 4.1 workshop pair
├── benchmarks/                   pipeline, source and UDF benchmarks
├── tests/                        pytest: unit + integration (one file per service)
├── docs/                         getting started, guides, deployment, troubleshooting
├── terraform/                    AWS deployment (VPC, S3, RDS, EMR)
├── terraform-databricks/         Databricks as a managed destination
├── .github/ · .gitea/            CI and release workflows
├── CLAUDE.md                     agent guide to the project
├── AGENTS.md                     always-in-context Spark 4.1 reference for agents
└── .claude/skills/               agent skills, loaded on demand (SDP first; the rest grew into lakehouse-skills)

In April I reframed it from a hobby into a composable reference architecture, the pieces you'd need for a Databricks-equivalent lakehouse with each one swappable, and in May a stripped-down, Spark Connect-first copy became open-lakehouse for team demos on AWS. The original stays the messier workbench, and it's still where new versions land first.

Some of it is unfinished, and I'd rather say so. The Terraform for AWS and Databricks has never been applied end to end, and the IoT and personal-data half of the plan is still a plan. But the thing I built it for, it does. AUTO CDC shipped in Spark 4.2 in July; by then I'd already run it on a Spark that didn't officially exist yet, against data designed to misbehave, on my own desktop, and I knew where its edges were.

All About Iceberg's File Format API: simplifying the engine/format interface and Parquet's multimodal future

lnc — Wed, 14 Oct 2026 05:21:01 +0000

What is the file format API?

Apache Iceberg traditionally exposed its data files through three libraries, one each for Parquet, Avro and ORC, and until this year every engine that spoke Iceberg had its own way of calling them. Spark had SparkParquetReaders, Flink had FlinkParquetReaders, the generic Java API had GenericParquetReaders, and each engine chose its own path through the format's read options, filter pushdown, split handling, encryption and metrics. The practical effect was that a feature added to one format for one engine (a new v3 type, a default value, a shredded variant column) tended to arrive for the other engines later or not at all, because the work had to be redone per engine.

The File Format API, which shipped in 1.11.0 and was announced in February, is the project's attempt to streamline all this: a single registry of "format models," each one the pairing of a file format with an engine's in-memory row type, and a single set of builders (ReadBuilder, FileWriterBuilder and its delete-file siblings) that every engine now goes through. It was also originally framed as the way to try out newer formats such as Vortex and Lance, which are fundamentally designed very differently than Parquet and meant specifically for vector work and multimodal data storage.

Note for Rust/Python users: this is a Java API. The registry lives in iceberg-core, so it reaches exactly the engines that link that jar and reuse its readers (Spark, Flink, Hive, the plain Java Record path). DuckDB, DataFusion via iceberg-rust, PyIceberg and Polars each implement the Iceberg spec with their own readers and leverage Arrow readers instead. For Trino, which is Java, it instead uses Iceberg's core for metadata and its own Parquet and ORC readers for data. DuckDB, DataFusion, Polars and PyIceberg all read into Arrow or something Arrow can hand over without copying, so the row-type axis collapses to one and a new format just needs an Arrow reader (Lance and Vortex both ship one). Java implementations have three row types and no shared memory format, which is why it needed a registry to play the role Arrow plays everywhere else.

Engine	Language	Reads Iceberg data files with	Goes through the registry?
Spark, Flink, Hive, plain Java API	Java	Iceberg's own readers from `iceberg-core` and the format modules	yes
Trino	Java	its own Parquet and ORC readers; Iceberg core for metadata only	no
PyIceberg	Python	its own readers on PyArrow	no
DataFusion (via iceberg-rust)	Rust	its own Arrow-based Parquet reader	no
DuckDB	C++	its own Parquet reader	no
Polars	Rust	Arrow	no

Table 1. Which engines the File Format API actually reaches.

Reducing Format Model x Reader Complexity

A file format is how bytes are laid out on disk: Parquet, Avro and ORC each have their own encoding, compression, statistics and footer, and Iceberg doesn't care which one a given data file uses as long as it can read it. An engine in-memory row type, which the project calls the object model, is how one row looks in memory once it's been read, and that depends on the engine, not the file: Spark wants an InternalRow, Flink a RowData, the plain Java API an Iceberg Record. The same Parquet file can be read into any of them, and the same InternalRow can be written out as Parquet or ORC. A format model is one cell in that grid: one format, one row type.

Here is the grid as it stands at 1.12.0, with a mark wherever a format model exists:

Row type (object model)	Parquet	Avro	ORC
`Record` (plain Java API)	yes	yes	yes
`InternalRow` (Spark, row at a time)	yes	yes	yes
`ColumnarBatch` (Spark, vectorized)	read only	no	read only
`ColumnarBatch` (Arrow)	read only	no	no
`RowData` (Flink)	yes	yes	yes

Table 2. The format-by-row-type grid. Each "yes" is one registered format model.

Scale this complexity further, per engine and per format:

After, through the File Format API's registry:

**The API makes the engine's row type pluggable over a fixed list of file formats, such that any engine can register its own representation and read every Iceberg table through the shared builders, while a new format remains a change inside the Iceberg project; and because the formats are fixed, multimodal support is arriving as new column types defined on the Parquet side (see FILE type that was shipped in parquet-format 2.14.0, VECTOR type currently in proposal stage) and mirrored into the next table spec.

Into the weeds

Everything in the API is built on one map, which makes it the ideal place to start digging into in the codebase to gain a better understanding. If you only want to read and write files through the registry, skip ahead to "Usage." Read this section to understand why the API behaves the way it does, and come back to it when something in a later section needs the detail.

The registry key

A map here is the ordinary programming kind: a lookup table from a key to a value. In FormatModelRegistry, the key is a pair (which file format, which row type) and the value is a format model, the object that knows how to read and write that one combination. Here is the method that puts entries into the map:

// core/src/main/java/org/apache/iceberg/formats/FormatModelRegistry.java (1.12.0)
public static synchronized void register(FormatModel<?, ?> formatModel) {
  Pair<FileFormat, Class<?>> key = Pair.of(formatModel.format(), formatModel.type());
  ...
  MODELS.put(key, formatModel);
}

The second half of the key, Class<?>, is the Java class of the row the engine wants back. For the plain Java API it is Iceberg's own Record; for Spark it is InternalRow (one row at a time) or ColumnarBatch (a block of rows); for Flink it is RowData. The Iceberg project calls this the object model.

[!note] Object model versus schema
The object model is the engine's in-memory shape for a row. It says nothing about the file and nothing about the table's schema. The schema says which columns a table has; the object model says what Java class a row of that table becomes once it is read.

The first half of the key, FileFormat, is an enum: a type whose allowed values are a fixed list written into the source code. You cannot add a value at runtime. This enum lives in iceberg-api, the module that holds Iceberg's public interfaces:

// api/src/main/java/org/apache/iceberg/FileFormat.java (1.12.0)
public enum FileFormat {
  PUFFIN("puffin", false),
  ORC("orc", true),
  PARQUET("parquet", true),
  AVRO("avro", true),
  METADATA("metadata.json", false);
  ...
}

Puffin and metadata appear because the enum names every kind of file Iceberg writes, not only data files. Puffin files hold statistics and deletion vectors; metadata.json is the table's root file. The three data formats are the ones in the middle: ORC, Parquet and Avro.

The two halves of the key are not the same kind of thing, and that difference organizes the rest of this post:

Half of the key	Type	Allowed values	Who can add one
Format	`FileFormat` enum in `iceberg-api`	the five names above	the Iceberg project, by changing the enum
Row type	any `Class<?>`	anything	any engine, by calling `register()`

Table 3. The two halves of the registry key.

The announcement says the registry "decouples engines from specific formats and allows new formats to be added without modifying engine code." That is true as far as engine code goes. What a new format has to modify is iceberg-api, the enum itself.

What happens on a read

A scan task is Iceberg's description of one file to read, including which format it is in. From the task to the rows, a read goes like this:

The engine asks the registry for a read builder, passing the task's format, its own row type, and the file.
The registry looks up the (format, row type) pair.
If the pair is registered, the model returns a builder. The engine sets the columns it wants, a filter, and the byte range of the file it was assigned, then calls build().
If the pair is not registered, the registry throws an error that names the pair it could not find.

The dotted lines draw the open-versus-closed distinction: one input to the lookup comes from a fixed list, the other is whatever the engine says it is.

How models get registered

When the registry class is first loaded, it tries to call a register() method on four classes by name: the generic models, and the Arrow, Flink and Spark ones. If one of those classes, or something it depends on, is not on the classpath (the set of jars the program was started with), the registry logs a line and moves on. The practical consequence is that the contents of the map depend on which Iceberg jars you ship. You will see this directly in the "Usage" section. Everything the project registers at 1.12.0 is an engine's row type against a format Iceberg already had (Table 2); none of it is a new format.

Usage

It would be a terrible user experience to construct format models yourself; so instead you'll ask the registry for a builder and configure it. A builder is an object you set options on one at a time and then finish with build(). The registry exposes two of them, and together they are the whole surface an engine touches:

Call	You pass	You set	You get back
`readBuilder(format, rowType, inputFile)`	the file's format, your row type, the file	columns to project, a filter, a byte range (split)	something you can iterate over, yielding rows of your type
`dataWriteBuilder(format, rowType, outputFile)`	the format to write, your row type, the output	the schema and partition spec	a writer; closing it yields a `DataFile`, the record Iceberg stores in its manifests, with column statistics already filled in

Table 4. The two registry calls an engine uses.

Where this code lives

If your experience of Iceberg so far is spark.sql("SELECT * FROM demo.orders") or a REST catalog behind a /v1/namespaces endpoint, the first question is where code like this even goes. It goes nowhere near either of those. There are three different places you can stand when you use Iceberg, and the File Format API only exists at the lowest one:

Where you stand	What you type	What Iceberg does for you	Sees the Format API?
An engine (Spark SQL, Flink SQL, Trino)	`SELECT`, `INSERT`, `MERGE`	catalog lookup, planning, file reads and writes, commits	No. The engine calls it for you.
A catalog client (the REST catalog, PyIceberg, a Spark catalog config)	load a table, list snapshots, commit metadata	metadata only; no data files are opened	No. Files are not touched at this level.
The Java library directly (`iceberg-core`, `iceberg-data`, `iceberg-parquet` on a classpath)	Java against `Table`, `Schema`, `FormatModelRegistry`	only what you call	Yes. This is where the registry is a public class you can call.

Table 5. Three levels of using Iceberg. The Format API is reachable only from the bottom one.

The program below is a plain Java main method with the Iceberg jars as dependencies. It is not a Spark job, it is not submitted to anything, and it does not need a catalog or even a table. It writes one Parquet file with Iceberg's own writer and reads it back with Iceberg's own reader, which is the single step every engine performs thousands of times per query and the only step the Format API is responsible for.

You would never run something like this in production. You run it when you want to see what the registry contains on a given classpath, or when you are writing a new format model and want the shortest possible loop between a change and a result. Here is the whole setup:

Make an empty directory with two files in it: pom.xml and src/main/java/demo/Main.java.
In pom.xml (Maven is Java's package manager, and this is its dependency list), declare the Iceberg modules you want on the classpath. The set is deliberately minimal: iceberg-api, iceberg-core, iceberg-data and iceberg-parquet, plus the Hadoop client jars Parquet needs and a logger. There is no iceberg-orc, and no Spark.
Run mvn -q compile exec:java. Maven downloads the jars from Maven Central, compiles Main.java, and runs it on the JVM in your terminal. The output below is what it prints.

The program

I ran this against Iceberg 1.12.0 from Maven Central on Java 17, with nothing timed. The program does four things:

Prints the FileFormat enum.
Reaches into the registry and prints what is in the map.
Writes three records through the write builder, then reads them back through the read builder.
Asks for a (format, row type) pair that cannot be there.

package demo;

import java.io.File;
import java.lang.reflect.Method;
import java.util.Map;
import org.apache.iceberg.DataFile;
import org.apache.iceberg.FileFormat;
import org.apache.iceberg.Files;
import org.apache.iceberg.PartitionSpec;
import org.apache.iceberg.Schema;
import org.apache.iceberg.data.GenericRecord;
import org.apache.iceberg.data.Record;
import org.apache.iceberg.encryption.EncryptionUtil;
import org.apache.iceberg.formats.FormatModelRegistry;
import org.apache.iceberg.io.CloseableIterable;
import org.apache.iceberg.io.DataWriter;
import org.apache.iceberg.types.Types;

public class Main {
  public static void main(String[] args) throws Exception {

    // 1. The format half of the key: a fixed list, nothing is added at runtime.
    System.out.println("== FileFormat enum values");
    for (FileFormat f : FileFormat.values()) {
      System.out.println("  " + f + " splittable=" + f.isSplittable());
    }

    // 2. Peek inside the registry (reflection, since the map is private).
    //    Shows what the jars on the classpath registered before main() ran.
    System.out.println("== registered format models on this classpath (core + data + parquet)");
    Method models = FormatModelRegistry.class.getDeclaredMethod("models");
    models.setAccessible(true);
    Map<?, ?> m = (Map<?, ?>) models.invoke(null);
    m.keySet().stream().map(Object::toString).sorted().forEach(k -> System.out.println("  " + k));

    // An Iceberg schema. Columns are matched by the numeric field id, not by name.
    Schema schema = new Schema(
        Types.NestedField.required(1, "id", Types.LongType.get()),
        Types.NestedField.optional(2, "name", Types.StringType.get()));

    // A local file to write to; an engine would get an object store path from FileIO.
    File out = new File(args.length > 0 ? args[0] : "/tmp/ffa-demo.parquet");
    out.delete();

    // 3a. Write. (PARQUET, Record.class) is the registry key; build() does the lookup
    //     and hands back the Parquet model's writer, or throws if the pair is missing.
    System.out.println("== write via FormatModelRegistry.dataWriteBuilder(PARQUET, Record.class, ...)");
    DataWriter<Record> writer =
        FormatModelRegistry.<Record, Object>dataWriteBuilder(
                FileFormat.PARQUET, Record.class,
                EncryptionUtil.plainAsEncryptedOutput(Files.localOutput(out)))
            .schema(schema)
            .spec(PartitionSpec.unpartitioned())  // required even without partitioning
            .overwrite()
            .build();

    // Record is Iceberg's generic row type; copy(name, value, ...) makes a row.
    GenericRecord rec = GenericRecord.create(schema);
    for (long i = 1; i <= 3; i++) {
      writer.write(rec.copy("id", i, "name", "row-" + i));
    }
    writer.close();

    // toDataFile() returns the manifest entry an engine would commit, statistics included.
    DataFile df = writer.toDataFile();
    System.out.println("  wrote " + df.path() + " format=" + df.format() + " records=" + df.recordCount()
        + " bytes=" + df.fileSizeInBytes() + " valueCounts=" + df.valueCounts());

    // 3b. Read. Same key, other direction. An engine would also set a filter and a split.
    System.out.println("== read via FormatModelRegistry.readBuilder(PARQUET, Record.class, ...)");
    try (CloseableIterable<Record> rows =
        FormatModelRegistry.<Record, Object>readBuilder(
                FileFormat.PARQUET, Record.class, Files.localInput(out))
            .project(schema)
            .build()) {
      for (Record r : rows) {
        System.out.println("  " + r);
      }
    }

    // 4. A valid format and a valid row type, but iceberg-orc is not on the
    //    classpath, so nobody registered the pair. The error names it.
    System.out.println("== ask for a model that is not registered (ORC with generic Record, no iceberg-orc on classpath)");
    try {
      FormatModelRegistry.readBuilder(FileFormat.ORC, Record.class, Files.localInput(out));
    } catch (IllegalArgumentException e) {
      System.out.println("  " + e.getMessage());
    }
  }
}

And its output:

== FileFormat enum values
  PUFFIN splittable=false
  ORC splittable=true
  PARQUET splittable=true
  AVRO splittable=true
  METADATA splittable=false
== registered format models on this classpath (core + data + parquet)
  (AVRO, class org.apache.iceberg.deletes.PositionDelete)
  (AVRO, interface org.apache.iceberg.data.Record)
  (PARQUET, class org.apache.iceberg.deletes.PositionDelete)
  (PARQUET, interface org.apache.iceberg.data.Record)
== write via FormatModelRegistry.dataWriteBuilder(PARQUET, Record.class, ...)
  wrote .../out.parquet format=PARQUET records=3 bytes=684 valueCounts={1=3, 2=3}
== read via FormatModelRegistry.readBuilder(PARQUET, Record.class, ...)
  Record(1, row-1)
  Record(2, row-2)
  Record(3, row-3)
== ask for a model that is not registered (ORC with generic Record, no iceberg-orc on classpath)
  Format model is not registered for format ORC and type interface org.apache.iceberg.data.Record

Reading the output top to bottom:

The map holds Parquet and Avro, each paired with the generic Record and with the position-delete row type, and nothing else.
Avro is present without being asked for, because its model ships inside iceberg-core.
ORC is absent because the iceberg-orc jar is not on the classpath, and the error names the exact pair the registry could not find.
The write returned valueCounts: the writer filled in the DataFile statistics without the program doing anything.
One thing the program cannot express at all is a request for a format outside the five names. There is no FileFormat.LANCE to pass, and converting the string "lance" to a FileFormat throws "Invalid file format: lance" before the registry is ever reached.

The row-type side is open. The format side is closed.

The same call, from Spark

To connect this back to the level you already know: when you run spark.read.table("demo.orders").show() with the Iceberg connector, Spark loads the table through the catalog, plans which data files each task should scan, and then, for every file, makes exactly the call the program made by hand. Here is the Spark connector's version at 1.12.0, from BaseRowReader in the iceberg-spark module:

// Same call as the program above, with Spark's row type and all the options set.
ReadBuilder<InternalRow, ?> reader =
    FormatModelRegistry.readBuilder(format, InternalRow.class, file);
return reader
    .project(projection)          // columns the query asked for
    .idToConstant(idToConstant)   // partition values, not stored in the file
    .reuseContainers()
    .split(start, length)         // this task's byte range
    .caseSensitive(caseSensitive())
    .filter(residual)             // what is left of the WHERE clause
    .withNameMapping(nameMapping())
    .build();

The format variable is the FileFormat recorded in the manifest entry for that file, and InternalRow.class is the row type Spark registered for itself in SparkFormatModels at startup. The query you type at the top level becomes, at the bottom level, one registry lookup per file. Everything between the two is planning.

Exploring pluggability

The extension point is relatively easy to grokk and doesn't take a ton of code. This section covers what an engine supplies, what the format model owns, and what "pluggable" therefore means in practice.

Here is how the generic Parquet model is registered, with each argument labeled:

// data/src/main/java/org/apache/iceberg/data/GenericFormatModels.java (1.12.0)
FormatModelRegistry.register(
    ParquetFormatModel.create(
        Record.class,          // the row type
        Schema.class,          // the engine schema type
        (icebergSchema, fileSchema, engineSchema) ->              // how to write a Record
            GenericParquetWriter.create(icebergSchema, fileSchema),
        (icebergSchema, fileSchema, engineSchema, idToConstant) ->  // how to read one
            GenericParquetReaders.buildReader(icebergSchema, fileSchema, idToConstant),
        new RecordVariantShreddingAnalyzer(),                      // see below
        (Function<Schema, UnaryOperator<Record>>) ignored -> Record::copy));  // how to copy a row

The engine supplies two functions. The writer function, given a schema, builds something that writes a row's values into Parquet. The reader function builds something that reads values out of Parquet into a row; its idToConstant argument is a map from column ids to values that come from metadata rather than from the file, mostly partition columns. ParquetFormatModel wraps those two functions in the builders every caller uses, and everything else lives in the model, once, for every engine:

The engine supplies	The format model owns
the row class and the engine schema type	schema conversion between Iceberg's schema and the file's
a writer function (row values into the file)	filter pushdown: skipping chunks of the file that cannot match
a reader function (file values into a row)	splits: reading one byte range of a large file
a shredding analyzer (which variant fields to shred)	encryption
a row-copy function	the statistics that end up in the `DataFile`
	the shredded write path itself

Table 6. Division of labour between an engine and a format model.

The engine never touches anything in the right-hand column.

[!note] The analyzer argument
Variant is Iceberg's type for semi-structured data such as JSON. Shredding is Parquet's technique of storing some of a variant's fields as ordinary columns so they can be filtered and read quickly. The Parquet model owns the shredded write path; each engine supplies an analyzer that says which fields to shred for its row type. There is one for Record here, and a Spark one and a Flink one in their modules. It is a small example of how the labour divides: the format owns the mechanism, the engine owns the choice.

So if you maintain an engine, or anything with its own idea of a row, this is your plugin point, and it is a good one. Register your row type against Parquet, Avro and ORC, and your engine reads any Iceberg table through the same builders Spark uses, without changing anything in the Iceberg project.

And the engines really do all go through it now, which is the part of the announcement that is simply true. Every reader in the 1.12.0 source that opens a data file (in Spark, Flink, the generic Java path, the MapReduce input format and the Arrow reader) gets its reader from the registry; none of them calls the old per-format Parquet.read, ORC.read or Avro.read directly any more. One implementation per format, shared by every engine, chosen by (format, row type).

Put that next to the dotted lines in the read diagram and "pluggable" comes into focus. What was scattered and is now unified is the engine side. What was fixed and is still fixed is the format side. That is not a flaw so much as a boundary, and the next section is about what sits on the other side of it.

Where formats come into play and the compatibility kit

This section answers two questions: how does a new file format get into Iceberg today, and what does it have to pass once it is there?

Adding a format: the Vortex example

A new format enters Iceberg the way the project's own showcase does. Vortex is a newer columnar file format, and its integration (issue 15416) has a pull request (#15915) that shows every step. The PR has been open since April, and the table shows where it lands:

Layer	What a new format needs there	Does the Vortex PR touch it?
`iceberg-api`	a new `FileFormat` value	yes, first file in the change list
a format module	the format model, readers and writers	yes, new `iceberg-vortex` module
each engine	a registration of its row type against the new format	yes, Spark readers and tests
the compatibility kit	an entry so the whole test matrix runs against it	not yet
the table spec	a fifth name in the `file_format` sentence	no

Table 7. Where a new file format has to land, and how far the Vortex PR gets.

The first file in the PR's change list is FileFormat.java, adding VORTEX to iceberg-api, the module with the strongest compatibility promises in the project; every change there is checked by a tool called Revapi and has to be acknowledged on purpose. What the PR does not touch is as telling as what it does. Iceberg's table spec is the document every implementation follows, in Java or any other language. It lists each data file in a manifest with a file_format field, and the spec at 1.12.0 defines that field as "String file format name, avro, orc, parquet, or puffin." The Vortex PR changes nothing in the spec. The Java enum is moving ahead of the spec text, which is a normal order of events in this project, but it means adding a format has two halves: the constant in iceberg-api, and the spec sentence that would need a fifth name before PyIceberg or iceberg-rust were obliged to know about it.

What a format has to pass

Behind the implementation sits the Technology Compatibility Kit (TCK). A compatibility kit is a test suite that a new component must pass to count as equivalent to the existing ones. Here, it is the set of tests a new format has to pass to be held to the same standard as Parquet. The announcement lists the TCK as in progress; the issue's author said in August that it is "more-or-less finished" apart from some v3 types and vectorized paths, which I'll take at face value and no further, since it's one comment.

The path for a format today is therefore:

Get into the enum.
Implement a format model.
Get every engine to register its row type against it.
Pass the TCK.

That is a project roadmap item, not a plugin. A closed enum is also how the DataFile metadata, the manifest field above, and every reader that switches on format stay consistent with one another, so I'd be surprised if it opened casually.

Inside the kit

Since the kit is what a format or an engine actually has to pass, it is worth looking at what it is, because it is smaller and more concrete than the name suggests.

The kit is two abstract test classes in the test folder of the iceberg-data module. Abstract means they are templates: they contain every test but leave a few blanks for an engine to fill in. ReadFormatModelTests covers reading, and BaseFormatModelTests extends it with writing. An engine joins by writing a small subclass that fills in the blanks:

What its row class is.
How to turn an Iceberg schema into its own kind of schema.
How to turn a generic Record into one of its rows.
How to compare two of its rows.
How to convert a single constant value.

Spark's subclass is about a page, Flink's about the same. Everything else is inherited from the template.

What gets inherited is every test, run for every file format the kit knows about, times every data generator (a small class that produces a schema and some random rows for it). The generators cover primitives, UUID, fixed, binary, decimals, a struct, a list and a map of primitives, nanosecond timestamps and variant. The tests cover round trips in both directions (including delete files), every read builder option, default values at every nesting, statistics in each collection mode, the metadata columns, schema evolution and name mapping.

The part I find most useful is that the kit is also where the exceptions get written down. Each engine's subclass declares the Iceberg types it cannot hold, and each format declares the features it is missing. Tests for those are skipped rather than failed, so the kit doubles as the honest table of who supports what:

	Cannot hold / is missing
Spark	`time`, `timestamp_ns`
Flink	`time`
Avro	filter pushdown, splits, column statistics, native encryption
ORC	reader-side defaults, variant, container reuse
Parquet	AES stream encryption

Table 8. Exclusions declared in the compatibility kit at 1.12.0.

The nanosecond timestamp I mentioned above is a one-line entry in Spark's exclusion set, which is the clearest statement anywhere in the repository that the gap is in the engine and not in the format or the API.

The kit was not in 1.11.0. It arrived as a series of separate PRs through the year, several from contributors outside the original design group, and the last open piece is vectorized reads.

That is what sits behind the "more-or-less finished" comment, and it also tells you how a new format gets in: add an entry for it to the kit's list of formats, run the whole matrix, and write down what it cannot do in the missing-features map. That list is the contract.

Implications for native multimodal in Iceberg

Multimodal data means images, audio, video and documents, plus the embedding vectors a model produces from them. If formats are fixed, then multimodal data has to come into Iceberg through the formats it already has, and that is exactly what is happening, one level down and one project over. Two new column types are being defined in the Parquet format spec first, and then mirrored into Iceberg's table spec as part of the v4 revision. (The table spec is versioned; v3 is current, and v4 is being drafted in the open PRs below.)

Both new types are Parquet logical types. A logical type is a label on top of a physical storage type that says how to interpret the bytes: a string is bytes labeled as UTF-8, a date is an integer labeled as days since the epoch.

FILE

FILE is a column type for "bytes that live somewhere." Parquet got a FILE logical type in parquet-format PR 585, merged in July and released in 2.14.0 in September. Iceberg's FILE type proposal (Talat Uyarer, Alex Stephen, Sung Yun and Gaurav Soni; spec in PR 17918) defines the Iceberg file type by reference to it. A value is a reference with optional fields:

Field	Meaning
`uri`	where the bytes live
`offset`, `size`	a byte range inside that object
`content_type`	what the bytes are
`checksum`	integrity check
`inline`	the bytes themselves, for small payloads

Table 9. Sub-fields of the proposed Iceberg file type.

So a small image can sit inline inside the data file, and a large one can be a byte range inside an external object. The table carries the reference, statistics on the sub-fields, and mappings for Parquet, Avro and ORC.

Whether an engine can actually go and fetch those bytes without holding cloud credentials is a separate question, handled on the catalog side with pre-signed URLs: links that carry their own temporary permission. PR 18080 adds that to the REST catalog spec, and PR 18079 teaches the readers to use the links the catalog hands back (both by Sung Yun, both open). That is where the design is still moving.

VECTOR

VECTOR follows the same pattern, a few months behind. parquet-format PR 624 proposes a VECTOR(D) logical type, meaning a list of exactly D numbers, on top of Parquet's existing list storage. Iceberg's vector type proposal with its spec PR 18416 defines vector(E, D), a fixed-length list of D numbers of element type E, and maps it onto the three formats Iceberg has:

Iceberg v4 type	Parquet	Avro	ORC	Rules
`vector(E, D)`	Parquet's standard nested list encoding, annotated `VECTOR(D)` (per parquet-format #624); element required, Parquet mapping for `E`	`array`, elements non-null	`array`, elements non-null	`E` in `int`, `long`, `float`, `double`; exactly `D` non-null elements; no NaN or infinities; no type promotion to or from `vector`; no lower/upper bounds

Table 10. The proposed vector type and its format mappings.

What the two proposals have in common

Read those two proposals next to the registry and the shape is clear. Neither one routes embeddings or media around Parquet. Both extend Parquet to carry them, and then teach Iceberg's schema, statistics and three existing format mappings about the new type. Neither needs a new FileFormat. That was the part I didn't expect going in: the announcement talks about new formats, and the multimodal column types are being written into Parquet, the format the new formats were meant to sit beside.

The File Format API helps here, but less than you might hope, and it is worth being exact about how much. What the Parquet model shares across engines is the schema conversion, splits, filters, encryption and statistics. Turning a vector column into a Spark row field or a Flink row field is still the job of each engine's own reader and writer function, the two functions from the registration code in "Exploring pluggability." So a new type is one change in the Parquet module plus one in each engine's value readers, not one change total. (Both spec PRs are open; if either is withdrawn, or Iceberg defines the type without reference to Parquet, this section is dated.)

A useful way to hold all of this: a column type has to clear three layers before you can use it from a given engine, and the File Format API only touches the middle one. The table spec has to define it; each file format has to store it; each engine has to have somewhere to put the value. Table 11 shows four types at 1.12.0, two shipped and two proposed:

Type	Table spec	Parquet	Avro	ORC	Spark	Flink
`timestamp_ns`	v3	yes	yes	yes	no	yes
`geometry`, `geography`	v3	yes	yes	no	4.1 and later	no
`file`	v4, proposed	released in parquet-format 2.14.0	mapping in the proposal	mapping in the proposal	no	no
`vector`	v4, proposed	proposed	mapping in the proposal	mapping in the proposal	no	no

Table 11. Where four column types stand across the three layers. A "no" in the engine columns means the engine has no in-memory type to hold the value, whatever the file can store.

The nanosecond timestamp row is a keen example. The spec has it, every format stores it, and Spark's own timestamp is microseconds, so Iceberg's converter has no case for it. No change to the registry could fix that, meaning unsupported types in an engine don't suddenly become supported by this new API. Spark is in the middle of fixing it from its own side: an open proposal, SPIP: Timestamps with nanosecond precision, adds nanosecond timestamp types to Spark SQL, the types are already on Spark's main branch, and a steady stream of PRs is teaching each function about them. Once a Spark release ships them, the Iceberg side is a converter case and a one-line removal from that exclusion set, so do expect this row to change.

State of the work and future

Everything above is a reading of the Java implementation at 1.12.0 and of spec PRs that are open on the day I'm writing. It says nothing about PyIceberg or iceberg-rust, which I didn't read, and it benchmarks nothing; the run exists to show what the registry contains and what it says when a model is missing, not how fast anything is. The Vortex PR could merge in any form, and the Java enum could grow a value before the spec sentence does, as that PR already shows. If the enum is opened in some more general way, the "formats are a project change" half of this is dated. The types half stands on its own, because the FILE and vector types are defined in Parquet regardless of what the Java registry does.

Piece	Status	Where to look
File Format API	shipped in 1.11.0; every engine in 1.12.0 reads through it	announcement, proposal
Compatibility kit	"more-or-less finished" per its author; vectorized reads open	issue 15415
Vortex as a new format	open PR, adds `VORTEX` to the enum	PR 15915
Parquet `FILE` type	released in parquet-format 2.14.0	PR 585
Iceberg `file` type (v4)	open spec PR	issue 17919, PR 17918
Pre-signed data access	open spec and reader PRs	PR 18080, PR 18079
Parquet `VECTOR` type	open proposal	PR 624
Iceberg `vector` type (v4)	open spec PR	issue 18297, PR 18416

Table 12. Status of every moving part, as of writing.

Keeping up with the Joneses: What your next step is

You have images or embeddings in Iceberg today. Keep the reference as a string and the embedding as a list, which the current readers handle, and plan a schema migration to file and vector when v4 is adopted rather than a storage migration to another format.
You want to follow the decisions. Watch the two Iceberg spec PRs, the Parquet VECTOR PR, and the two pre-signing PRs, because that is where the capability is being decided.
You are an engine author with your own row type. Register it. The API is for you, it works, and the run above is the whole recipe.
You want a new format in Iceberg. Start from the TCK issue and the Vortex PR, and budget for it as a contribution to the project, because that is what it is.

An Idiot's Guide to Spark Connect Clients

lnc — Wed, 14 Oct 2026 05:20:55 +0000

Spark Connect is a new concept for many of us, especially given the long history of how managing Spark clusters has traditionally worked. Spark Connect in a nutshell separates your application from Spark: your code becomes a small client, Spark runs as a server somewhere else, and the two talk over the network. The primary benefit of Connect for most users is getting rid of the headaches around SparkContext, installing a package from your notebook, your Python version not mattering, or a typo failing on the line you typed it, which Connect very kindly obfuscates for us. Honestly, it's a pretty overdue upgrade for Spark as an engine and a pretty intuitive direction: decouple client sessions from the server, and make the server and its configuration as dead easy as possible for your customers.

The mindset shift away from having your PySpark setup share the same process as your Spark driver is not a super hard jump, but it does mean there are some adjustments to make in how you set things like configurations or write jobs. This becomes even more intense once you get into the meat of Spark Connect, which is the language agnosticism and running multiple clients and servers, particularly for those of us dabbling in Rust or Go.

So this guide, written for idiots by an idiot, starts with a tour of how Spark itself works and how that changes with a Connect setup. From there it's practical: how to run a server, how to connect to it from Python and other languages, what changes in your code, and how to do some basic debugging. Note that after the inclusion of session tracing in 4.3, and hopefully an eventual Spark Connect Gateway, the latter will be much, much easier.

Every example comes from spark-demos, where each one is a runnable script, and I ran every snippet here against Spark 4.2.0, the current release.

How Spark works

Spark lets you write code as if you were working with one table on one machine, and then runs it across however many machines you have; Connect's whole trick is moving part of the machinery that makes that possible somewhere else, so it's worth having a picture of that machinery first. A running Spark application has a few kinds of pieces:

The driver is the brain. It holds your SparkSession, works out what your code actually means, decides how to run it, and coordinates everything else. Inside it:
- the analyzer, which resolves the tables and columns you named, using the catalog (the registry of what tables exist and what's in them);
- the Catalyst optimizer, which rewrites your query into a cheaper equivalent (skipping columns you never use, filtering before joining, and so on);
- the planner, which turns the optimized query into physical operations;
- the schedulers, which cut that physical plan into stages (at every point where data has to be reshuffled between machines) and each stage into tasks, one per slice (partition) of the data.
The cluster manager hands out machines: Spark's own standalone master, YARN, or Kubernetes.
Workers are the machines (or containers) the cluster manager gives you.
Executors are the processes on those workers that actually run tasks, read and write data, hold cached data, and exchange data with each other during a shuffle.

Following one line of code through all of that:

orders.filter(orders.total > 20).groupBy("region").count().show()

Step	What happens	Where
1	Your code describes the query: filter, group, count	your program
2	That becomes an unresolved logical plan: a tree of operations naming tables and columns that haven't been checked yet	the driver
3	The analyzer checks `orders`, `total` and `region` against the catalog and fills in types	the driver
4	Catalyst optimizes the plan (here it reads only the two columns it needs)	the driver
5	The planner picks physical operators and generates code for them	the driver
6	The schedulers split the work into stages (before and after the shuffle `groupBy` needs) and tasks	the driver
7	Executors run the tasks against their slices of the data	the executors
8	The counts come back to the driver, and on to you	the driver, then your program

Nothing actually runs until step 8 is needed; until you call something that wants a result (show(), count(), writing a table), Spark is only building the plan, which is what people mean by lazy evaluation. Classic Spark does run step 3 straight away (it checks your names the moment you define a DataFrame), and that small difference ends up mattering a lot once Connect is involved, so keep it in your back pocket for the debugging section.

The detail that matters most for this guide is where the driver runs. In classic PySpark (in the default "client" mode) the driver runs inside your program: your Python process starts a Java virtual machine next to it and talks to it through a bridge called Py4J, and that JVM is the driver, with the analyzer, the catalog connection, Catalyst and the schedulers all living in it.

Your code and Spark's brain share a process, a machine, a set of installed libraries, and a fate, such that if one of them crashes it takes the other down with it.

What the shared driver costs you

Sharing a process is fine when it's one person on one laptop, and it causes a surprising number of problems as soon as more than one person (or more than one program) uses the same Spark:

You need Java and all of Spark on every machine that runs your code, including your laptop, a CI runner, or a web service that just wants to run one query.
Your application and Spark are upgraded together. Upgrading Spark means changing the environment every job runs in, whether the job was ready or not.
Everyone's code shares the driver. One badly behaved notebook can take the driver down for everyone attached to it.
Only JVM languages and Python can talk to Spark, because they're the ones that can live inside or next to the JVM.
Credentials go everywhere the driver goes, so catalog passwords and storage keys end up on every machine that runs a job.

Spark Connect fixes all of these with one change, which is that it splits the driver in two. The part you write against, the DataFrame API, stays in your program; everything that thinks (the session, the analyzer, the catalog, Catalyst, the schedulers) moves into a long-running Spark Connect server, with a network connection in between. Your program builds step 2, the unresolved plan, as a small protobuf message, sends it over gRPC and gets results back as Arrow, and steps 3 to 8 all happen on the server.

Piece by piece, this is what moves and what stays put:

Piece	Classic PySpark	Spark Connect
Your code and the DataFrame API	your process	your process (the client)
Building the unresolved plan	the driver JVM, via Py4J	the client, as a protobuf message
`SparkSession` and its state (temp views, settings, artifacts)	the driver, in your process	the server
Analyzer and catalog connection	the driver, in your process	the server
Catalyst, physical planning, code generation	the driver, in your process	the server
Stage and task scheduling	the driver, in your process	the server
Cluster manager, workers, executors	the cluster	the cluster, unchanged
Results	Py4J and a local socket into your process	Arrow batches over gRPC
Catalog and storage credentials	wherever the driver runs	the server only

The cluster itself doesn't change at all, and executors run tasks exactly as they did before. What changes is that your program no longer contains a driver, so it doesn't need Java, it can be written in any language that speaks gRPC and Arrow, and it can be upgraded on its own schedule rather than the server's. Connect first appeared in Spark 3.4, and since 4.0 the server has shipped in the standard Spark download.

Pretty much every confusing thing about Connect is this split showing up somewhere you didn't expect it. explain() shows you a plan your laptop didn't make, because Catalyst is on the server; a typo in a column name fails late, because the analyzer is on the server and only sees your plan once you ask for a result; sparkContext is gone, because it belonged to the driver. Anything that used to live in your program (the JVM, your files, your Python version, your credentials) now lives on the other side of a network connection, and most of the rest of this guide is working through the consequences of that.

Connecting from Python

From Python it's two steps: install the thin client (pyspark-client),

pip install pyspark-client

and create your session with .remote(...) instead of .master(...):

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

spark = SparkSession.builder.remote("sc://localhost:15002").getOrCreate()
print(spark.version)            # the *server's* Spark version

orders = spark.createDataFrame(
    [("metro", "pizza", 24.0), ("metro", "sushi", 40.0), ("urban", "pizza", 18.0)],
    "region string, item string, total double",
)
orders.groupBy("region").agg(F.round(F.avg("total"), 2).alias("avg_ticket")).show()

Everything after the session line is ordinary PySpark, the same code you'd run in classic mode. Your program builds the plan and the server optimizes and runs it, so if you call .explain() you're reading the server's plan (your laptop doesn't have an optimizer anymore).

Before and after: setting up a session

A real classic session usually does a lot more than pick a master; it pulls in packages, sets SQL extensions, turns on Hive support and quiets the logs. A typical one looks something like this:

# Before: Spark Classic. Everything is configured by the program that runs the driver.
spark = (
    SparkSession.builder
    .master("spark://my-cluster:7077")
    .config("spark.jars.packages", "org.apache.spark:spark-sql-kafka-0-10_2.13:4.2.0")
    .config("spark.sql.shuffle.partitions", "64")
    .enableHiveSupport()
    .getOrCreate()
)
spark.sparkContext.setLogLevel("WARN")

Under Connect that one block splits in two, because some of those settings describe the driver, and the driver isn't yours anymore. Anything the server reads once when it starts (packages and JARs, spark.sql.extensions, catalog implementations, Hive support) moves to the command that starts the server:

# After, part 1: the server owns its JVM, so it gets the packages and static settings.
./sbin/start-connect-server.sh \
  --master spark://my-cluster:7077 \
  --packages org.apache.spark:spark-sql-kafka-0-10_2.13:4.2.0

and the client keeps only what can change per session:

# After, part 2: the client names the server and sets runtime options.
spark = (
    SparkSession.builder
    .remote("sc://localhost:15002")
    .config("spark.sql.shuffle.partitions", "64")   # runtime settings still work from the client
    .getOrCreate()
)

The log level moves to the server as well (its conf/log4j2.properties), since the logs are the server's now. The sneaky part is that Connect doesn't stop you from setting a startup-only setting from the client; it warns and carries on without it. Asking for a package from the client gets you this, and no package:

UserWarning: Failed to set spark.jars.packages to Some(org.apache.spark:spark-sql-kafka-0-10_2.13:4.2.0)
due to [CANNOT_MODIFY_CONFIG] Cannot modify the value of the Spark config ...

enableHiveSupport() does the same thing (with CANNOT_MODIFY_STATIC_CONFIG) and hands you a session without Hive, which you might not notice until a table you expected isn't there. The rule of thumb I've landed on is that if a setting describes how the server starts, it belongs wherever the server starts.

You can also leave the URL out of your code entirely by setting SPARK_REMOTE, and a plain SparkSession.builder.getOrCreate() will connect to it:

export SPARK_REMOTE="sc://localhost:15002"

Before and after: the rest of your code

Most DataFrame code doesn't change at all. What does change is anything that reached into the driver, and accumulators are a good example:

# Before: an accumulator, counted on the executors and read on the driver
bad_rows = spark.sparkContext.accumulator(0)
orders.foreach(lambda row: bad_rows.add(1) if row.total is None else None)
print(bad_rows.value)

# After: an Observation, collected alongside a query you were running anyway
from pyspark.sql import Observation

obs = Observation("quality")
checked = orders.observe(obs, F.count(F.when(F.col("total").isNull(), 1)).alias("bad_rows"))
checked.write.mode("overwrite").saveAsTable("orders_checked")
print(obs.get["bad_rows"])

I'd argue the Observation is the better pattern anyway, since the metric rides along with a write you were already doing instead of needing its own pass over the data. The rest of the common swaps, side by side:

Before (Classic)	After (Connect)
`spark.sparkContext.setLogLevel("WARN")`	set the level in the server's `conf/log4j2.properties`
`df.rdd.zipWithIndex()`	`df.zipWithIndex()` (4.2+)
`spark.read.json(rdd_of_strings)`	`spark.read.json(df)` on a string column (4.2+)
`sc.accumulator(0)`	`df.observe(Observation(...), ...)`
`sc.addPyFile("lib/geo.py")`	`spark.addArtifact("lib/geo.py", pyfile=True)`
`sc.broadcast(lookup)`	a broadcast join, `F.broadcast(lookup_df)`, or a literal column
`spark.newSession()`	`SparkSession.builder.remote(url).create()`
`spark-submit job.py`	`python job.py` with `SPARK_REMOTE` set

All of that assumes there's a server sitting at localhost:15002, which brings us to running one.

Running a Connect server

There are three ways to get one, roughly in order of effort.

1. Inside your own Python process, for trying things out. Full PySpark can start a Connect server in-process:

spark = SparkSession.builder.remote("local[*]").getOrCreate()

This needs full pyspark (not pyspark-client) and Java installed, because the server still has to run somewhere; it's the Connect code path without a separate server, which makes it great for tests and for learning.

2. From a Spark download, on your machine or a server. Download Spark from spark.apache.org (you'll need Java 17 or newer) and start the server that ships with it:

./sbin/start-connect-server.sh        # listens on port 15002
./sbin/stop-connect-server.sh         # when you're done

By default it runs Spark locally on that one machine; to have it run jobs on a cluster instead, point it at the cluster's master:

./sbin/start-connect-server.sh --master spark://my-cluster:7077

One thing that surprised me is that by default the server listens on every network interface, so anything that can reach your machine can reach the server the moment it starts. On a laptop on shared Wi-Fi, restrict it with --conf spark.connect.grpc.binding.address=127.0.0.1. (You'll see spark.connect.grpc.binding.host in plenty of examples, including some of mine; it isn't a real setting, and Spark silently ignores it.)

3. In Docker. The official apache/spark image includes Spark, Java and Python; add PyArrow and pandas on top, since Python UDFs use Arrow by default as of Spark 4.2:

FROM apache/spark:4.2.0-scala2.13-java21-python3-ubuntu
USER root
RUN python3 -m pip install --no-cache-dir pyarrow pandas
USER spark

Then start the server with --wait, which keeps it in the foreground (what a container needs):

services:
  spark-connect:
    build: .
    entrypoint:
      - /opt/spark/sbin/start-connect-server.sh
      - --wait
    ports:
      - "127.0.0.1:15002:15002"   # Spark Connect
      - "127.0.0.1:4040:4040"     # Spark UI

The image's Python is 3.10, which matters for your client (more on that below).

For production, Connect servers also run on Kubernetes (the Apache Spark Kubernetes Operator deploys one as an ordinary Spark application) and, since Spark 4.2, inside YARN. The repo's topology guide walks through six ways to deploy one and when each makes sense.

Before you share a server with other people, know that a single server holds many sessions, each isolated from the others but all sharing the server's memory and executors; if people need stronger isolation than that, they need separate servers.

Running a job

There's no separate "submit" step for a Connect job, because your program is the job, and running it sends the work:

export SPARK_REMOTE="sc://localhost:15002"
python my_pipeline.py

(SparkSession.builder.getOrCreate() picks up SPARK_REMOTE from the environment, so the same script runs against whichever server the variable names.) A few other routes are worth knowing about.

An existing job, almost unchanged. spark.api.mode switches a classic job to Connect: Spark starts a Connect server for the job and runs your code as its client. Set it in the session builder and keep your existing master:

spark = (SparkSession.builder
         .master("spark://my-cluster:7077")
         .config("spark.api.mode", "connect")
         .getOrCreate())

This is the gentlest way to try Connect on real work, since you can move one job, watch it, and move it back if something breaks. The documentation also shows passing it to spark-submit as --conf spark.api.mode=connect; on 4.2.0 that failed for me with TypeError: 'JavaPackage' object is not callable while the builder setting worked, so for now, set it in code.

A declarative pipeline. Tell the spark-pipelines CLI which server to use with --remote:

spark-pipelines run --spec spark-pipeline.yml --remote sc://localhost:15002

Don't rely on SPARK_REMOTE here; the CLI ignores it and quietly starts a local server of its own. The spark-pipelines command comes with full PySpark, and with only the thin client it's the other way round: run python -m pyspark.pipelines.cli run --spec spark-pipeline.yml, which has no --remote option and connects to whatever SPARK_REMOTE names.

Code your job depends on. If your functions import your own modules, the server doesn't have them, so ship them with the session using addArtifact:

spark.addArtifact("lib/geo.py", pyfile=True)        # importable by your UDFs
spark.addArtifact("model.tar.gz", archive=True)     # unpacked on the executors
spark.addArtifact("lookup.csv", file=True)          # copied as is

JARs for catalogs and extensions are the exception; those have to be installed on the server.

What you need locally

Much less than before, which is kind of the point:

You need	Why
Python, the same minor version as the server's	Python functions you send run in the server's Python. The 4.2 client needs Python 3.10 or newer
`pyspark-client`	The thin client
Its dependencies: pyarrow, pandas, numpy, grpcio	They come with it; this is most of the install
No Java, no Spark download, no `SPARK_HOME`	That all lives on the server

The thing actually worth getting right is which package you install, because there are four with nearly identical names:

Package	What you actually get	Java needed locally?
`pyspark`	Full PySpark: about 470 MB, most of it some 280 JAR files	yes
`pyspark[connect]`	Full PySpark plus the Connect extras	yes
`pyspark-connect`	A tiny package that installs full `pyspark` (and makes Connect its default)	yes
`pyspark-client`	The thin client: about 15 MB, no JARs	no

pyspark-connect sounds like the Connect client and isn't; the thin client is pyspark-client. Don't judge by the size of your environment either: with its dependencies a pyspark-client environment is about 330 MB against about 780 MB for pyspark[connect], because pyarrow, pandas and numpy are most of it, so what the thin client really saves you is Java, not disk space. To check which one you actually have:

import importlib.util, pathlib, pyspark

print((pathlib.Path(pyspark.__file__).parent / "jars").exists())  # thin client: False
print(importlib.util.find_spec("py4j") is not None)               # thin client: False

If either prints True, you're on full PySpark. And if you have a full Spark installed elsewhere on the machine, clear SPARK_HOME and PYTHONPATH before you start, or they can quietly shadow the thin client.

Clients in other languages

Because the client only has to build a plan and read Arrow, it doesn't have to be Python, or even a JVM language, and the same server takes all of these:

Language	Client	Latest release
Python	`pyspark-client` (in `apache/spark`)	4.2.0
Scala and Java	`spark-connect-client-jvm` (in `apache/spark`)	4.2.0
Go	`apache/spark-connect-go`	v0.1.0
Swift	`apache/spark-connect-swift`	0.7.0
Rust	`apache/spark-connect-rust`	4.2.0
Anything with JDBC	`spark-connect-client-jdbc`, URL `jdbc:sc://host:15002`	since 4.1

They all start the same way, with a connection string. In Scala:

val spark = SparkSession.builder().remote("sc://localhost:15002").getOrCreate()
spark.range(10).filter("id % 2 = 0").show()

In Go:

spark, err := sql.NewSessionBuilder().Remote("sc://localhost:15002").Build(ctx)
df, err := spark.Sql(ctx, "SELECT 'apple' AS word, 123 AS count")
df.Show(ctx, 100, false)

(At the time of writing, the Go client's only release, v0.1.0, can't be fetched with go get because of a module-path mismatch; go get github.com/apache/spark-connect-go@master works.)

Rust is where it gets fun for me, because results arrive as Arrow and can go straight into a native Rust tool. This program from the repo has Spark aggregate five million rows down to seven and then hands them to Polars for local work:

let spark = SparkSession::builder().remote("sc://localhost:15002").get_or_create()?;
let e = |c: Column| c.expression().clone();   // aggregate expressions, unwrapped

let summary = spark
    .range(5_000_000)?
    .with_column("bucket", col("id") % lit(7))
    .group_by([col("bucket")])
    .agg(vec![e(f::count(col("id")).alias("n"))]);

let batches = summary.collect_record_batches()?;   // Arrow RecordBatches, ready for Polars

That division of labour, with Spark doing the part that needs a cluster and your language of choice doing the rest, is the part of Connect I find most interesting. The clients aren't equally complete yet, though (the Python and JVM ones cover the most, the newer ones less), so check that the one you pick supports what you need before you build on it.

Debugging

Most Connect problems come from forgetting that there are now two sides, so when something goes wrong I work through these in roughly this order.

1. Am I talking to the server I think I am? Print spark.version first, every time, since nothing in sc://localhost:15002 says which Spark is supposed to answer. I learned this one the hard way: an old Spark 4.1 server was still sitting on port 15002 on my machine when I pointed newer tools at it, and it answered, misunderstood a request it had never heard of, and gave me an error convincing enough that I wrote it up as a Spark bug. (The full story is in Hosting an Entire Open Lakehouse on Port 8080.) Two related traps:

a = SparkSession.builder.remote("sc://server-a:15002").getOrCreate()
b = SparkSession.builder.remote("sc://server-b:15002").getOrCreate()
assert b is a    # getOrCreate() returned the existing session; use .create() for a new one

and connection strings that look fine but aren't; the part after the port has to be empty, and parameter values can't contain /:

sc://localhost:15002/;user_agent=myapp/setup
PySparkValueError: [INVALID_CONNECT_URL] ... The path component '/;user_agent=myapp/setup' must be empty.

2. Read the error's name. Connect errors carry an error class in square brackets, and a handful of them cover most of what you'll hit early on:

Error	What it means	What to do
`JVM_ATTRIBUTE_NOT_SUPPORTED`	You used something that lived in the local JVM: `spark.sparkContext` (and so everything reached through it, like `sparkContext.setLogLevel`), `_jvm`, `df.rdd`	Use the DataFrame replacement: `df.zipWithIndex()`, `spark.read.json(df)`, `df.observe(...)` for counters, `addArtifact` for files
`PYTHON_VERSION_MISMATCH`	Your Python and the server's differ (e.g. 3.12 vs 3.10)	Match your Python minor version to the server's
`ModuleNotFoundError` inside a UDF	The server's Python is missing a package your function needs	Ship it with `addArtifact`, or install it on the server
`UNSUPPORTED_LOCAL_CONNECTION_STRING`	You called `.create()` with `local[*]`	Use `.getOrCreate()` for local
`INVALID_CONNECT_URL`	The connection string is malformed	Empty path, no `/` in parameter values

One that doesn't raise at all is enableHiveSupport(), which prints a warning and gives you a session without Hive, so read your warnings.

3. Errors arrive late. Classic Spark checks your DataFrame the moment you define it (that step 3 from earlier); Connect doesn't send anything to the server until it needs an answer, so a typo in a column name fails at the show() or write rather than on the line with the typo, and a try/except around the definition catches nothing. The same thing has two quieter consequences:

a.createOrReplaceTempView("brands")
first = spark.table("brands")
b.createOrReplaceTempView("brands")
first.count()     # Classic: rows of a    Connect: rows of b

A DataFrame built on a temp view reads whatever the view points to when it runs, and a Python function that reads a variable sees that variable's value when it runs, so give temp views unique names and pass values into functions explicitly.

4. Look at the server's side. The server has the Spark UI (port 4040 by default), where you can see the jobs your client triggered along with a Connect tab listing every client session and request, and it has its logs (in logs/ under the Spark directory, or docker logs for a container). df.explain() shows you the plan the server actually built.

5. Check your client isn't newer than your server. The supported direction is an older client talking to a newer server, and the other way round can fail in odd ways: a 4.2 client can't even build a DataFrame from a three-row Python list on a 4.1 server, because the 4.1 server reports a size limit as 3221225472b and the 4.2 client chokes on the b.

6. Let a script check for you. The repo includes a doctor that checks the client you installed, whether the server is reachable, its version, and whether Python functions actually run on it:

python tools/doctor.py --remote sc://localhost:15002 --expect-version 4.2

The whole guide, on one page

Spark splits big data work across many machines.
Spark Connect moves the driver out of your program; your code becomes a client that sends plans to a server.
Use it with pip install pyspark-client and SparkSession.builder.remote("sc://host:15002").
Get a server in-process with local[*], from a Spark download with start-connect-server.sh, or from the apache/spark Docker image (plus PyArrow).
Send a job by running your program; ship your own modules with addArtifact.
Locally you need Python matching the server's version and pyspark-client, and nothing else.
Other languages (Scala, Go, Rust, Swift, JDBC) use the same server and the same connection string.
When it breaks, check which server answered, read the error's name, remember errors arrive late, and look at the server's side.

Nearly all of it comes back to the one change at the start, which is that the driver moved out of your program; once you know which side of the network a thing lives on, most of Connect stops being surprising.

Spark's Modules, Seen Through Spark Connect

lnc — Wed, 14 Oct 2026 05:20:49 +0000

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:

Over Spark Connect	Status
DataFrame operations, Spark SQL, catalog operations, temp views	Supported
Structured Streaming	Supported
Python, Pandas, and Arrow UDFs; UDTFs	Supported
`pyspark.ml` pipelines (fit / transform)	Supported, via a server-side model cache
`SparkContext`, RDDs, broadcast variables, accumulators	Not supported

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.

Package	What it is
`pyspark-client`	The thin client: pure Python, no JDK, remote only (`sc://...`)
`pyspark[connect]`	Full PySpark plus Connect extras; can also start a local server
`pyspark-connect`	A small shim that depends on full `pyspark` (so not thin)
`pyspark`	Classic Py4J by default; talks to a Connect server via `.remote()` with the extras installed

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":

Statement (abridged)	Result on 4.2.0
`SELECT * FROM <metric view>`	`INTERNAL_ERROR` at planning
Selecting a measure without `MEASURE()`	`INTERNAL_ERROR` at planning
Joining a metric view on a dimension	`INTERNAL_ERROR` at planning
`MEASURE()` over an ordinary table	`INTERNAL_ERROR`
`CREATE VIEW ... WITH METRICS` without `LANGUAGE YAML`, and a metric view defined over another metric view	`INTERNAL_ERROR`

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.

Concept	Python API	Behavior
Materialized view	`@dp.materialized_view`	Batch table, recomputed each run
Streaming table	`@dp.table`	Incremental ingestion with managed checkpoints
Flow	`@dp.append_flow`	Named writer into a streaming table
AUTO CDC flow (4.2)	`dp.create_auto_cdc_flow(...)`	Applies a change feed as SCD Type 1

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:

Client → server	Result
4.2.0 → 4.2.0	Everything works
4.1.2 → 4.2.0	Works; newer APIs simply don't exist in the old client
4.1.2 → 4.1.2	Works
4.2.0 → 4.1.2	`createDataFrame` from a local list fails; `GetStatus` returns `UNIMPLEMENTED`

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

Module	Verdict	Reachable over Connect	What I'd do
Spark SQL	Built for it	Yes	Default choice; push logic here
Declarative Pipelines	Built for it	Yes	Use for multi-table pipelines; AUTO CDC for change feeds
Structured Streaming	Rebuilt	Yes	Micro-batch for stateful work
Real-Time Mode	Rebuilding	Yes from 4.2 (stateless)	Stateless only until 4.3; check Python gaps
MLlib (`spark.ml`)	Rebuilt	Yes (since 4.0)	Fine for existing algorithms
Python UDFs	Rebuilt	Yes	`@arrow_udf` first; Arrow transport is default in 4.2
GraphX	Left behind	No	Migrate to GraphFrames
SparkR	Left behind	No	Migrate to `sparklyr`
DStreams, `spark.mllib`	Left behind	No	Migrate to DataFrame APIs

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.

AUTO_CDC: Turning an event log into current state with Spark Declarative Pipelines

lnc — Wed, 14 Oct 2026 05:20:43 +0000

The question an event log can't answer cheaply

The dataset I use for most of my pipeline demos is a simulated ghost kitchen (a delivery-only restaurant operation), and like most operational data it arrives as an append-only log. Every order produces a lifecycle of events, each stamped with a time and a per-order sequence number, and a day of it is about a million events. The generator deliberately injects a few percent of late and duplicated events, because real streams do that.

Logs like this are excellent at answering what happened and awkward at answering what operations people actually ask: which orders are live right now, and where is each one?

The usual answer is to recompute it. Group the whole history by order, take the latest event for each, drop the cancelled ones, and repeat on every refresh, paying for the entire history every time and getting it subtly wrong whenever "latest" means latest to arrive rather than latest to happen.

The better answer depends on noticing something that sounds obvious once it's said: an append-only lifecycle log is already a change data capture feed. Each event is a change to the current state of one entity (the order), keyed by order id, sequenced by when it happened, and a cancellation is simply a delete.

Here is an illustrative order (simplified; the real sequence is described below), first as the lifecycle log records it, then as the change feed it implies:

seq	event_type	as a change to `orders_live`
1	`order_created`	UPSERT order 1042, status = `order_created`
2	`kitchen_started`	UPSERT order 1042, status = `kitchen_started`
3	`order_ready`	UPSERT order 1042, status = `order_ready`
4	`driver_picked_up`	UPSERT order 1042, status = `driver_picked_up`
5	`driver_ping`	UPSERT order 1042, status = `driver_ping`
6	`delivered`	UPSERT order 1042, status = `delivered`

An order that ends in order_cancelled produces a DELETE as its last change instead, and disappears from the table. Once the log is seen this way, "current status of every order" stops being an aggregation problem and becomes the problem of applying a change feed to a table, which is exactly what AUTO CDC was built to do.

My argument is that AUTO CDC is worth adopting less because it saves code (though it does, a lot) than because it moves the hard parts of change application into the engine, and leaves you with a small, explicit contract: a sequence column that is unique and never null, a target that genuinely supports row-level operations, and deletes expressed as events. Most of what goes wrong with hand-rolled CDC is the engine's side of that contract done badly; most of what can still go wrong with AUTO CDC is your side of it.

What AUTO CDC is, and where it stands

AUTO CDC began as APPLY CHANGES in Databricks' Delta Live Tables and was contributed to Apache Spark under its own SPIP (SPARK-56249), as part of Spark Declarative Pipelines. In Python it's create_auto_cdc_flow.

Release	AUTO CDC status
Spark 4.2.0 (July 2026)	Shipped. SCD Type 1 via the Python / Spark Connect API
Spark 4.3.0 (release candidate)	Adds SCD Type 2 (history with validity ranges)
main branch (5.0.0-SNAPSHOT)	Ongoing work under SPARK-56249

Everything I tested was SCD Type 1 (the target holds only the current state per key, and updates overwrite in place), on a build of the main branch from late May 2026, before 4.2 was released.

Applying a change feed correctly means doing several things at once:

matching each change to its row by key
keeping the version with the highest sequence per key, not the last to arrive
honouring deletes, but only when the delete is the latest thing that happened to that key
staying idempotent when a streaming source redelivers after a failure

The hand-rolled version is usually a foreachBatch with windowed deduplication feeding a MERGE, and it tends to get the cross-batch cases wrong without throwing errors. A late update that lands in the micro-batch after a newer one quietly overwrites it, and nobody notices until the numbers disagree with something else.

The code

The projection from lifecycle log to change feed is ordinary Spark. This is the core of what my test harness does: every event becomes an UPSERT of the order's current status, a cancellation becomes a DELETE, and the sequence is made unique per order.

feed = events.select(
    F.col("order_id"),
    F.col("event_type").alias("status"),
    F.col("location_id").cast("int").alias("location_id"),
    F.col("ts_seconds").alias("event_ts"),
    F.when(F.col("event_type") == "order_cancelled", F.lit("DELETE"))
     .otherwise(F.lit("UPSERT")).alias("op"),
    # unique per-key ordering: event time, broken by per-order sequence
    (F.col("ts_seconds") * F.lit(100) + F.col("sequence")).alias("seq"),
)

Then the pipeline itself is three declarations: a streaming source, a target streaming table, and the flow.

from pyspark import pipelines as dp

@dp.table(name="order_cdc_source")
def order_cdc_source():
    return spark.readStream.schema(ORDER_SCHEMA).json("file:///data/order_cdc/*.json")

dp.create_streaming_table("ice.gold.orders_live")

dp.create_auto_cdc_flow(
    target="ice.gold.orders_live",
    source="order_cdc_source",
    keys=["order_id"],                  # how a change finds its row
    sequence_by="seq",                  # highest sequence per key wins, whatever the arrival order
    apply_as_deletes="op = 'DELETE'",   # which events delete the row
    except_column_list=["op", "seq"],   # bookkeeping columns to leave out of the target
    stored_as_scd_type=1,
)

The engine handles what the hand-rolled version gets wrong: deduplication across micro-batches by sequence_by, the generated upsert and delete, exactly-once checkpointing, and the auxiliary state table that remembers the latest sequence per key across runs.

That table is what lets a late event arriving in a later batch resolve correctly.

	Recompute from the log	AUTO CDC
Cost per refresh	re-aggregates the entire history	processes only new events
Out-of-order events	"latest" by arrival is wrong	resolved by `sequence_by`
Cancellations	a row every query must remember to filter	the row is deleted
Code	bespoke `foreachBatch` + dedup + `MERGE`	three declarations

Running it on messy data

I ran the order-status projection on a sample of the ghost-kitchen stream: a little under five thousand orders and roughly 125,000 of their lifecycle events, including the injected late and duplicate ones. AUTO CDC collapsed them into one row per live order:

`gold.orders_live`	rows
`delivered`	4,601
`driver_ping` (mid-delivery)	84
`order_ready`	3
`order_created`	1
`kitchen_started`	1
`driver_picked_up`	1
total live orders	4,691
cancelled orders removed by `apply_as_deletes`	308

To check it, I recomputed the same table from scratch the naive way and compared the two in both directions:

w = Window.partitionBy("order_id").orderBy(F.desc("seq"))
naive = (feed.withColumn("rn", F.row_number().over(w)).filter("rn = 1")
             .filter("op != 'DELETE'").select("order_id", "status"))
pairs = spark.table("ice.gold.orders_live").select("order_id", "status")
match = naive.subtract(pairs).count() == 0 and pairs.subtract(naive).count() == 0
# match = True

They matched exactly.

It's worth being precise about what that proves: it compares the two columns the table exists to answer, as sets, rather than every column byte for byte.

Two details of the run also deserve honesty, because they're the kind of thing that makes a demo look cleaner than it was. The sample was drawn by sorting order ids and taking the first five thousand, and because the raw data contains events with a null order id (which Spark sorts first), the sample was really 4,999 orders.

And the projection joined each event to its order's creation event, so orders with duplicated creation events fanned out into duplicate rows in the feed. AUTO CDC handled both without complaint, since duplicate copies of an event carry the same sequence value and resolve to the same row, but a pipeline that quietly processes inputs it wasn't meant to receive is a reminder to look at your inputs before trusting your outputs.

The second scenario was the textbook one: keeping a dimension current as kitchens relocate, rename, and close. These are the exact changes streamed into silver.dim_locations_current, on top of an initial load of four locations at sequence 1:

changes = [
    {"id": 2, "city": "Los Angeles - Arts District", "op": "UPSERT", "seq": 5},
    {"id": 1, "city": "San Francisco - SoMa",        "op": "UPSERT", "seq": 4},
    {"id": 3, "city": "Chicago",                      "op": "DELETE", "seq": 6},   # kitchen closed
    {"id": 2, "city": "Los Angeles (STALE)",          "op": "UPSERT", "seq": 3},   # arrives late, older
]

id	after AUTO CDC	why
1	San Francisco - SoMa	rename at seq 4 applied
2	Los Angeles - Arts District	rename at seq 5 kept; the stale seq 3 edit ignored
3	(no row)	closure at seq 6 deleted it
4	Austin	unchanged

A nightly full reload would eventually reach the same names, but it copies every unchanged row to get there and has no way at all to express that a location no longer exists.

That's the more general lesson: an append-only or aggregate pipeline can only add facts, so "this was cancelled" becomes another row every downstream query has to remember to filter. In a CDC pipeline a terminal business event can be the delete itself, and the same mechanism propagates hard deletes for compliance.

Edge cases, and how Type 1 resolves them

I put SCD Type 1 through a battery of adversarial cases, and it behaved correctly in every one. The behaviours are worth knowing precisely, because this is where intuition tends to be wrong. These are the cases from my stress harness, with the events in arrival order; | separates micro-batches.

Case	Events for one key	Result
Upsert to a key that never existed	UPSERT@1	row inserted
Delete of a key that never existed	DELETE@1	no-op
Several updates in one batch	UPSERT@1, UPSERT@2, UPSERT@3	row has the @3 values
Stale delete	UPSERT@1 \| UPSERT@3 \| DELETE@2	row stays, with the @3 values; the update is newer
Stale insert after a delete	UPSERT@3 \| DELETE@5 \| UPSERT@2	row stays deleted
Delete arrives before an older insert	DELETE@5 \| UPSERT@2	row stays deleted
Resurrection	UPSERT@1 \| DELETE@2 \| UPSERT@3	row present, with the @3 values
Volume, out of order	500 keys, each UPSERT@1, @3, @2 shuffled across 5 batches, plus DELETE@5 for 100 of them	exactly 400 rows, all with the @3 values
Tie (probe, not pass/fail)	UPSERT@5 "Boston", UPSERT@5 "Denver" in one batch	one of them, chosen arbitrarily

Your side of the contract

The sequence. A null in the sequence_by column fails the micro-batch with AUTOCDC_MICROBATCH_VALIDATION.NULL_SEQUENCE rather than guessing, which is the right behaviour, and means the column has to be non-null by construction.

Ties are worse, because they don't fail: two changes to the same key with the same sequence in one micro-batch are resolved arbitrarily, as the programming guide itself warns. A coarse timestamp is the classic way to produce ties, so sequence on something strictly increasing per key, like a log offset or commit LSN. My composite of timestamp times 100 plus the per-order sequence works only because no order in this data has more than 99 events; it's a local trick, not a general recipe. The ties that did occur in my data were all injected exact duplicates, the one case where an arbitrary choice is harmless.

The target. AUTO CDC needs a connector that implements Spark's DataSource V2 row-level operation interface (SupportsRowLevelOperations), and "supports MERGE" is not the same thing:

Target	Result in my testing
Plain Parquet	rejected immediately: `AUTOCDC_TARGET_DOES_NOT_SUPPORT_MERGE`
Delta	did not work: Delta's table implementation doesn't expose `SupportsRowLevelOperations`, despite supporting MERGE elsewhere
Iceberg, Hadoop catalog	worked
Iceberg, JDBC (Postgres) catalog	worked
Iceberg, Hive Metastore catalog	worked
Iceberg, Gravitino REST catalog	worked
Iceberg, Glue catalog (emulated)	worked
Iceberg, Unity Catalog OSS REST	failed: the catalog is read-only
Any ephemeral / in-memory catalog	unsuitable: the aux state table must persist across restarts

Two smaller surprises: the target carries a hidden __spark_autocdc_metadata column that shows up in SELECT *, and Iceberg's own changelog is batch-only in open-source Spark for now, so you can't stream one Iceberg table's changes straight into AUTO CDC as a source.

A correction. I originally reported that running an AUTO CDC pipeline through the spark-pipelines CLI failed at registration with InvalidPlanInput: RELTYPE_NOT_SET, and called it a bug. It wasn't.

The CLI connects to the default Spark Connect endpoint, and an old Spark 4.1 Connect server was still listening there, so my pipeline was being sent to a server that had never heard of AUTO CDC. Against a server of the right version the CLI works correctly. If you see that error, check what's actually listening on sc://localhost:15002 before anything else.

What changes

With a hand-rolled pipeline, the question you're left with is whether you've correctly implemented cross-batch deduplication, sequenced deletes, and idempotent merges, and the honest answer is usually that you won't know until something disagrees. With AUTO CDC that part belongs to the engine, which is tested against exactly the cases people get wrong.

The question becomes whether you've given it a sequence that is unique and never null, a target that really supports row-level operations, and a delete for everything that should be allowed to disappear. That's a much better question to be left with, and the first step is often not a new data source at all; it's noticing that the event log you already have has been a change feed all along.

From Pickle to Arrow: A Decade of Moving Data Between Spark's JVM and Python

lnc — Wed, 14 Oct 2026 05:19:50 +0000

Arrow's real contribution to PySpark isn't speed at any one API but a shared memory format at every boundary between the JVM and Python; each release removed one conversion at one boundary, what each one left behind explains the next, and the conversion that remains now lives inside Spark's own engine rather than at the Python edge.

That crossing is the oldest problem in PySpark, and the most persistent attempt to fix it started in February 2016, when Wes McKinney (the creator of pandas) filed a Spark ticket with a modest title: SPARK-13534, "Implement Apache Arrow serializer for Spark DataFrame for use in DataFrame.toPandas." It asked for one method to stop using pickle.

It took two years to ship. And over the decade that followed, the same idea went through every other door between Spark's JVM and Python: UDFs, partition-level APIs, the network protocol of Spark Connect, custom data sources, until in Spark 4.1 a Python function could receive Spark's data as an Arrow array and never touch pandas at all, and in Spark 4.2 the Arrow paths finally became the defaults.

The wall

PySpark is not a Python reimplementation of Spark. It's a Python client for a JVM engine. When you write df.filter(col("x") > 5).groupBy("y").count(), Python builds a query plan and hands it to the JVM, which optimizes, distributes, and executes it; Python never touches the data, which is why DataFrame code runs at the same speed in PySpark and Scala.

The trouble starts the moment your own code needs the data: a UDF, toPandas(), createDataFrame(pandas_df), a custom data source. Then data has to physically cross from the JVM into a Python process, and for most of Spark's history the only way across was pickle, one row at a time:

Every row pays four serialization steps and a function call. For a UDF like return x * 1.1, the arithmetic is nanoseconds and the machinery around it is microseconds, three orders of magnitude of overhead wrapped around almost no work.

That was PySpark from 2015 to 2018, and the same crossing existed at every door between the two runtimes, on both the driver and the executors:

Why the fix was a memory format, not a faster pickle

Arrow was conceived in late 2015 by people who had hit the same wall from different directions. McKinney had spent years fighting pandas' memory model; his 2017 essay "Apache Arrow and the 10 Things I Hate About pandas" lists the grievances, including a rule of thumb that pandas users needed "5 to 10 times as much RAM as the size of your dataset."

Jacques Nadeau, building a query engine at Dremio, saw every data system inventing its own internal format and paying to translate between them at every boundary. Marcel Kornacker, who created Impala at Cloudera, saw the same copying across the Hadoop ecosystem. They met at Strata NYC in September 2015, and Arrow was announced as an Apache top-level project in February 2016, skipping the incubator entirely, which is unusual and reflected how many companies wanted the same thing.

The idea was not a faster serializer. It was to agree on the bytes themselves: a language-independent, columnar, in-memory layout specified down to byte alignment, null bitmaps, and buffer order, so that two Arrow-aware systems can hand each other data without converting it.

Within a process, or with shared memory, that's genuinely zero-copy; across a socket it's still a copy, but with no per-row decoding on either side, because the bytes on the wire are already the bytes in memory.

Columnar mattered as much as standard. In a row layout, values from one column are interleaved with everything else in the row:

row layout:     [id=1, name="Alice", age=30] [id=2, name="Bob", age=25] [id=3, name="Carol", age=35]
columnar:       id:   [1, 2, 3]
                name: ["Alice", "Bob", "Carol"]
                age:  [30, 25, 35]

With columns stored contiguously, a 64-byte cache line holds eight useful int64 values instead of one row's mixed fields, and SIMD instructions can process a run of values at once.

The database literature had already shown what this buys: the MonetDB/X100 work (CIDR 2005) argued that pushing batches of around a thousand column values through each operator, instead of one row at a time, is an order of magnitude or two faster. Arrow's batch-oriented design was built around exactly that model.

Boundary one: the driver (2018)

The first door was the smallest. DataFrame.toPandas() collected results to the driver by pickling every row and rebuilding each one as Python objects before pandas could wrap them, and it was notoriously slow. McKinney's ticket, filed on February 28, 2016, three weeks after the first commit to the Arrow repository, asked to replace that path. It stayed open for over a year. Bryan Cutler, at IBM's Spark Technology Center, resolved it on June 23, 2017, and it reached users in Spark 2.3.0 in February 2018.

Two years, for one method.

# Spark 2.3 name, later renamed; on by default since Spark 4.2
spark.conf.set("spark.sql.execution.arrow.pyspark.enabled", "true")

The reverse direction, createDataFrame(pandas_df), got the same treatment (SPARK-20791). What made this matter wasn't the one method; it was the plumbing Cutler built to do it (Arrow converters on the JVM side, pyarrow integration on the Python side), which every later boundary reused.

What it removed: pickle for collecting data to the driver. What it left: everything that happens on executors. UDFs still crossed row by row.

Boundary two: UDFs, in batches (2018)

The second door was the one that mattered for performance, and it came from outside Databricks. The proposal, SPARK-21190 ("SPIP: Vectorized UDFs in Python"), was filed by Reynold Xin, but most of the implementation was done by Li Jin, an engineer at the quantitative investment firm Two Sigma and an Arrow committer, on top of Cutler's serializer. It shipped in the same Spark 2.3.0 release. Instead of calling your function once per row, Spark would send a batch of rows (10,000 by default, spark.sql.execution.arrow.maxRecordsPerBatch) as an Arrow RecordBatch and call your function once per batch, with a pandas Series:

# Before: once per row, pickle both ways
@udf(returnType="double")
def add_tax(price):
    return price * 1.1 if price else None

# After: once per batch, a pandas Series in and out
@pandas_udf("double")
def add_tax(price: pd.Series) -> pd.Series:
    return price * 1.1

Li Jin's post introducing the feature in October 2017 reported speedups from around 3x on simple arithmetic to around 100x on a cumulative-probability function, and the spread is the point: the simpler the function, the more of its cost was the crossing, and the more complex the function, the more it also benefits from running as vectorized NumPy instead of Python bytecode.

My own measurements, eight years and many releases later, land in the same shape. At 50 million rows, a pandas UDF did the arithmetic workload in 0.79 seconds against 5.20 for a pickled UDF, and a normal CDF in 0.82 against 6.18; on a string transform the gap was narrower, 2.51 against 6.22, for a reason that turns out to matter later.

What it removed: per-row serialization and per-row function calls. What it left: a rewrite (your function had to speak pandas), and a conversion. Every batch arrived as Arrow and was turned into pandas before your code saw it, and the result was turned back:

What actually crosses the socket

Every boundary from here on sends the same thing over the socket: Arrow's IPC stream format, a schema message followed by record-batch messages, each framed like this:

┌─────────────────────────────────────────────────────────────┐
│ Continuation indicator: 0xFFFFFFFF (4 bytes)                │
│ Metadata length (4 bytes, little-endian int32)              │
│ Flatbuffer metadata (Message type, schema or batch header)  │
│ Padding to 8-byte boundary                                  │
│ Message body (data buffers, padded to 8-byte multiple)      │
└─────────────────────────────────────────────────────────────┘

Inside the body, each column is a set of plain buffers: a validity bitmap with one bit per value, the values themselves as a contiguous typed array, and, for variable-length types like strings, an offsets array into one contiguous UTF-8 buffer. For four rows of (name: string, age: int32):

Validity bitmap (age):  [1, 1, 0, 1]      → 0b1011 (bit-packed)
Data buffer (age):      [30, 25, 0, 35]   → 16 bytes (4 × int32)
Offsets (name):         [0, 5, 8, 13, 18] → 5 × int32
Data buffer (name):     "AliceBobCarolDavid" (contiguous UTF-8)

No per-value type tags, no escaping, no framing between values.

Strings are where Arrow and pandas differ most: in Arrow, a million strings are one buffer and an array of offsets, while in traditional pandas they're a million separate Python objects.

Boundary three: letting code see Arrow (2022 to 2025)

For four years, every batch API in PySpark handed you pandas, whether you wanted pandas or not.

The first API that let user code see Arrow directly was mapInArrow, added by Hyukjin Kwon, a long-time PySpark maintainer at Databricks (SPARK-37227, with the Python side in SPARK-37228), in Spark 3.3.0 in June 2022. It passes whole partitions as an iterator of pyarrow.RecordBatch objects, and your function yields RecordBatches back:

import pyarrow as pa
import pyarrow.compute as pc

def add_tax_arrow(batches):
    for batch in batches:
        taxed = pc.multiply(batch.column("price"), pa.scalar(1.1))
        yield pa.RecordBatch.from_arrays([batch.column("id"), taxed],
                                         names=["id", "price_with_tax"])

df.mapInArrow(add_tax_arrow, schema="id long, price_with_tax double")

Grouped operations were the obvious remaining gap, and Enrico Minack closed it with applyInArrow (SPARK-40559), which shipped in Spark 4.0.0 in May 2025. Where applyInPandas turns each group into a pandas DataFrame, applyInArrow hands you the group as a pyarrow.Table:

def normalize(table: pa.Table) -> pa.Table:
    totals = table.column("total")
    mean, std = pc.mean(totals).as_py(), pc.stddev(totals).as_py()
    normalized = (pa.array([0.0] * len(totals)) if not std
                  else pc.divide(pc.subtract(totals, pa.scalar(mean)), pa.scalar(std)))
    return pa.Table.from_arrays([table.column("order_id"), totals, normalized],
                                names=["order_id", "total", "normalized"])

orders.groupBy("location_id").applyInArrow(
    normalize, schema="order_id long, total double, normalized double")

The same idea reached the other direction in 4.0 too: the new Python Data Source API (SPARK-44076) lets a custom source written in Python yield pyarrow.RecordBatch objects instead of tuples (SPARK-48493), so data read by Python code can enter Spark as Arrow instead of being built up one Python tuple at a time.

I haven't benchmarked the grouped or data-source paths myself, so there are no numbers for them here; the mechanism is the same as everywhere else, and the benefit should be largest where the pandas conversion is most expensive, which is strings.

What it removed: the forced detour through pandas, for anyone willing to write Arrow code. What it left: that condition. Most PySpark code is plain @udf functions that nobody is going to rewrite.

Boundary four: the network (2023)

Spark Connect started as a proposal from Martin Grund at Databricks in June 2022 (SPARK-39375): separate the client from the server, so an application talks to Spark over gRPC instead of living inside the driver's JVM. Once there's a network between you and the engine, the format results travel in stops being an optimization and becomes the protocol.

The answer was Arrow.

The client sends unresolved logical plans as Protocol Buffers; the server optimizes and runs them and streams the results back as Arrow record batches over the same channel. That's what makes a JVM-free Python client possible at all: the pyspark-client package is a 1.6 MB wheel of pure Python. The same columnar format that moves data between executors and Python workers now moves data between a remote cluster and a laptop.

Boundary five: the UDFs nobody rewrote (2023 to 2026)

Every boundary so far helped people who changed their code. Arrow-optimized Python UDFs (SPARK-40307, Spark 3.5.0, September 2023) took the opposite approach: keep your plain @udf exactly as written, and swap the transport underneath it. Batches cross as Arrow; Python unpacks each batch and still calls your function once per value.

Xinrong Meng's 2023 Databricks post measured about 1.6x over pickle for a single UDF and 1.9x for three chained ones on a 32 GB dataset, and my benchmark agrees in spirit: useArrow=True took the arithmetic workload from 5.20 to 2.23 seconds and the CDF from 6.18 to 2.94. Real, and cheap to get, but nothing like the pandas UDF numbers, because the per-row function call is still there.

That call is my benchmark's ceiling.

Turning this on by default took two tries. It was merged for Spark 4.0 (SPARK-48516) and reverted in March 2025 before the release; it finally shipped as the default in Spark 4.2, along with Arrow for toPandas() and for Python UDTFs, and a minimum PyArrow of 18.0. On 4.2, then, every existing @udf gets Arrow transport without anyone touching it.

It's also where I made the mistake that best illustrates how confusing this boundary is. My benchmark's "Arrow" variant of the plain UDF flipped spark.sql.execution.arrow.pyspark.enabled, which governs toPandas() and createDataFrame() and has nothing to do with UDFs. The setting that actually moves a UDF onto Arrow is spark.sql.execution.pythonUDF.arrow.enabled, or useArrow=True on the decorator. So my "Arrow" row was the pickled UDF measured twice, and its near-identical result was an accidental A/A test.

Three settings are named for Arrow, and they control three different boundaries.

What it removed: pickle, for everyone, with no code changes. What it left: the function call per row, which is the part no transport format can fix.

Boundary six: Arrow as the execution format (2025)

The last step turned Arrow from the way data travels into the way your code computes. Ruifeng Zheng's SPARK-53014, "Make Arrow UDF public," shipped in Spark 4.1.0 on December 16, 2025, almost ten years after McKinney's ticket. An Arrow UDF receives pyarrow.Array values and returns one, and it's meant to be written with pyarrow.compute, whose functions are C++ kernels operating directly on Arrow buffers:

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

@arrow_udf("double")
def add_tax(price: pa.Array) -> pa.Array:
    return pc.multiply(price, pa.scalar(1.1))

@arrow_udf("string")
def clean_email(email: pa.Array) -> pa.Array:
    return pc.utf8_trim(pc.utf8_lower(email))

The four UDF types differ only in where the data changes format:

UDF type	Data path	Format conversions
Pickle `@udf`	JVM → pickle → Python value → pickle → JVM	pickle both ways, per row
Arrow-optimized `@udf`	JVM → Arrow → Python value → Arrow → JVM	Arrow ↔ Python objects, per row
`@pandas_udf`	JVM → Arrow → pandas → Arrow → JVM	Arrow ↔ pandas, per batch
`@arrow_udf`	JVM → Arrow → your function → Arrow → JVM	none on the Python side

In my benchmark, @arrow_udf was the fastest Python option everywhere, but the margin over pandas UDFs depended almost entirely on the data type. On arithmetic and the CDF it was only 5 to 9% faster; on the string transform it was 1.6x faster at both 50M and 100M rows (1.55 seconds against 2.51 at 50M).

The split comes down to strings. A numeric pandas Series is a NumPy array underneath, so converting from Arrow is cheap; a string column in traditional pandas is an array of Python objects, so converting means building millions of them. pyarrow.compute skips all of that and works on the one contiguous buffer.

What it removed: the last conversion on the Python side. What it left: Spark's side.

Taken together, the choice today comes down to what your function can receive:

The pipe itself

Everything above changes what goes through the socket. Hyukjin Kwon's SPARK-51688, shipped in Spark 4.1.0, changes the socket: instead of TCP over loopback, which runs every byte through the full TCP/IP stack even though it never leaves the machine, the JVM and Python workers can talk over a Unix domain socket, which goes straight through a kernel buffer.

It's opt-in (spark.python.unix.domain.socket.enabled, whose default comes from the PYSPARK_UDS_MODE environment variable) in 4.1 and 4.2, and that's unchanged on the 4.3 branch, which instead adds another transport option: pipelined, asynchronous transfer of UDF batches between the JVM and Python (spark.python.udf.pipelined.enabled, off by default).

Spark 1.x:  pickle format  + TCP loopback        slow format over a slow pipe
Spark 2.3:  Arrow format   + TCP loopback        fast format over a slow pipe
Spark 4.1:  Arrow format   + Unix domain socket  fast format over a faster pipe (opt-in)

One boundary at a time

As of Spark 4.2, Arrow is the format at every place data crosses between the JVM and Python:

Boundary	Arrow's role	Default
`toPandas()` / `createDataFrame()`	replaces pickle for driver data collection	yes, since 4.2
`@udf` (plain Python UDFs)	replaces pickle transport	yes, since 4.2
`@pandas_udf`, `mapInPandas`, `applyInPandas`	batch transport, converted to pandas	always
`@arrow_udf`, `mapInArrow`, `applyInArrow`	transport and execution format	always
Spark Connect results	the result protocol	always
Python Data Source API	source data format	when the source yields RecordBatches

The last boundary

What's left is on Spark's side of the wall. Inside the JVM, Spark still executes on its own row format (UnsafeRow, from the Tungsten project), and Arrow exists at the edges, so every crossing still begins and ends with a conversion between Spark's rows and Arrow's columns. Engines like Velox and DataFusion show that a query engine can execute on Arrow-format data directly; Spark doesn't, and closing that gap is a much bigger change than any of the boundaries above.

The 4.3 branch, which hasn't been released yet (its first release candidate failed its vote on September 6, and a second, put to a vote on October 13, drew -1 votes over critical fixes that still need backporting, so there's no date), has two small signs of movement at exactly this edge. One is an Arrow format for Spark's in-memory cache (ArrowCachedBatchSerializer, SPARK-57268), part of a proposal to make Spark better on a single machine: the first time Arrow shows up as a storage format inside Spark itself rather than at a door.

The other goes around the boundary instead of through it: experimental transpilation of Python UDFs into Spark expressions (SPIP SPARK-54783), so a simple enough Python function never has to cross into Python at all. It's off by default, classic-mode only, and requires ANSI mode, but it's the one approach that could get under my benchmark's ceiling, because it removes the per-row Python call rather than making the trip to it cheaper.

The pandas side is moving too, more slowly than you'd expect. pandas 3.0, released in January 2026, made a PyArrow-backed string type the default, which should shrink exactly the string-conversion cost that separates pandas UDFs from Arrow UDFs. But PySpark doesn't support pandas 3 yet, not even on the 4.3 branch: it still pins pandas below 3.0, and the umbrella ticket for supporting it (SPARK-55139) is open.

When Wes McKinney filed SPARK-13534 in 2016, he was asking for one method to stop pickling. Ten years later, every door he could have named uses Arrow, the defaults finally say so, and the conversion that remains is the one Spark performs on itself, between its own rows and everything outside it.

That's the last boundary, and it was never a Python problem.

Most PySpark Tuning Advice Starts in the Wrong Place

lnc — Wed, 14 Oct 2026 05:19:44 +0000

When I benchmarked PySpark's UDF types earlier this year, I ran the same three workloads (trivial arithmetic, a string transform, a normal CDF) at five scales, from 100 thousand rows to 100 million. At 100 million rows, the default row-at-a-time Python UDF took about 10 seconds for arithmetic that a built-in function finished in under a fifth of a second. At 100 thousand rows, the same UDF was the fastest Python option I tested.

Both results are true, and between them they're most of what you need to know about PySpark performance: the cost is almost never in your function. It's in how many times, and in what shape, your data crosses from the JVM into Python, and whether that crossing is large enough to matter at all.

My argument is that there is really one PySpark performance technique, applied at different places: cross the JVM/Python boundary less often, move less across it, or stop paying for a JVM where you don't need one. The changes worth making are ordered by how much of that crossing they remove, and the biggest one, not crossing at all, is also the most often skipped.

The history of how Arrow came to carry that crossing is in Arrow in PySpark, 2016 to 2026, and the full benchmark is in Stop Blaming Serialization for Slow Python UDFs. This post is the practical version: what to change, in what order, and what each change actually removes.

Where the time goes

PySpark is a Python client for a JVM engine. DataFrame operations build a plan in Python (through Py4J, which only ever carries control messages, never bulk data) and the JVM executes it; df.filter(col("x") > 5).groupBy("y").count() never puts a row into Python, which is why it runs at the same speed as Scala. The data only leaves the JVM when your own code needs it:

Everything outside arrows ① and ② happens at native speed. Every change below makes those two arrows smaller, rarer, or unnecessary, and I've ordered them by how much they took off the arrows in what I measured, or, where I haven't measured, by what they remove.

Don't cross at all: built-ins and SQL UDFs

The largest win isn't a better UDF; it's no UDF. Built-in functions run inside the JVM as Catalyst expressions, fused by whole-stage code generation into tight loops, and never serialize anything.

What surprised me more is that SQL UDFs get the same treatment. Spark 4.0 added SQL user-defined functions (SPARK-46057), and their bodies are inlined into the logical plan, so the optimizer can see straight through them:

CREATE FUNCTION classify(score INT) RETURNS STRING
RETURN CASE
    WHEN score > 90 THEN 'excellent'
    WHEN score > 70 THEN 'good'
    ELSE 'needs_improvement'
END;

In my benchmark (median wall-clock time, 50M rows, Spark 4.1.0 through Spark Connect on one machine), the SQL UDF was indistinguishable from the built-ins, and every Python option cost several times more:

	arithmetic	string	normal CDF
built-in functions	0.107 s	0.486 s	0.176 s
SQL UDF	0.095 s	0.477 s	n/a
fastest Python UDF (`@arrow_udf`)	0.725 s	1.545 s	0.762 s
default Python UDF (`@udf`, pickle)	5.199 s	6.216 s	6.183 s

So the first question for any UDF is whether it needs to be Python. Plenty of "we need a UDF" logic is a CASE expression, a string function, or a date calculation that already exists in pyspark.sql.functions. Python earns its place when you need a library (scikit-learn, a parser, a C extension) or logic that genuinely can't be expressed in SQL. Everything after this point is about paying less for that choice.

Cross once per batch, not once per row

If it has to be Python, the next biggest lever is how often Spark calls your function. A regular @udf is called once per row; @pandas_udf and @arrow_udf are called once per batch of rows. That difference showed up as a hard ceiling in my runs: every row-at-a-time UDF, whatever it computed, topped out at roughly 8 to 10 million rows per second, which is roughly the rate at which CPython can be entered and exited. The batched UDFs weren't bound by it.

50M rows	arithmetic	string	normal CDF
`@udf` (pickle, per row)	5.199 s	6.216 s	6.183 s
`@udf(useArrow=True)` (Arrow transport, still per row)	2.228 s	4.430 s	2.942 s
`@pandas_udf` (per batch, `pandas.Series`)	0.789 s	2.512 s	0.819 s
`@arrow_udf` (per batch, `pyarrow.Array`)	0.725 s	1.545 s	0.762 s

Two things in that table decide what to change. Moving a per-row UDF onto Arrow transport helps, 1.4x to 2.4x here, and on Spark 4.2 you get it for free: 4.2 finally flipped spark.sql.execution.pythonUDF.arrow.enabled (and its UDTF counterpart) to true by default, after the same change was merged for 4.0 and reverted before release (SPARK-48516).

But rewriting to a batched UDF is worth several times more again, because it removes the per-row call rather than making the trip to it cheaper.

Between the two batched options, the choice mostly comes down to strings: @arrow_udf was only 5 to 9% faster than @pandas_udf on numeric work, but about 1.6x faster on the string transform, because pandas has traditionally stored strings as individual Python objects while Arrow keeps them in one contiguous buffer.

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

@arrow_udf("string")
def add_suffix(x: pa.Array) -> pa.Array:
    return pc.binary_join_element_wise(pc.utf8_upper(x), pa.scalar("_SUFFIX"), "")

One warning, because I learned it the expensive way: spark.sql.execution.arrow.pyspark.enabled is not the UDF setting. It governs toPandas() and createDataFrame(), and it's the one my benchmark originally toggled for its "Arrow" case, which is why I once reported that Arrow transport changes nothing for UDFs. It was the same pickled UDF, measured twice. (On 4.2 that setting is also on by default, so the confusion matters less for new code than for anyone reading old tuning advice.)

And the scale caveat from the opening is real. At small sizes, fixed costs dominate, and the ordering can flip:

arithmetic, median seconds	built-in	`@pandas_udf`	`@udf(useArrow=True)`	`@udf` (pickle)
100K rows	0.073	0.181	0.212	0.161
10M rows	0.074	0.299	0.568	1.092
100M rows	0.147	1.400	4.188	10.139

If your job touches a few hundred thousand rows, rewriting its UDFs is unlikely to be what makes it faster.

Don't produce the data: Python data sources with pushdown

After calling Python less, the next thing to remove is data that never needed to exist. Built-in sources like Parquet already push filters down and skip row groups that can't match; for most of Spark's history, custom sources that could do the same had to be written in Java or Scala. Spark 4.0 added the Python Data Source API (SPARK-44076), and Spark 4.1 added filter pushdown for it (SPARK-51271), so a source written in Python can now turn WHERE status = 'active' into a query parameter, a database predicate, or a skipped file block, and never send the rest.

When I went back over my own notes on this feature to write this post, the example I'd written was wrong, and checking it against Spark's source turned up something I hadn't known: pushdown for Python sources is off by default. It's governed by spark.sql.python.filterPushdown.enabled, an internal setting added in 4.1 whose description says enabling it costs "running Python worker one additional time during planning"; without it, your pushFilters() is never called.

The method also has a less obvious contract than I'd assumed. It returns the filters Spark must still apply after the scan, not the ones you accepted, and each filter's attribute is a column-path tuple:

from pyspark.sql.datasource import DataSourceReader, EqualTo

spark.conf.set("spark.sql.python.filterPushdown.enabled", "true")  # off by default

class ApiReader(DataSourceReader):
    def __init__(self, options):
        self.url = options.get("url")
        self.pushed = []

    def pushFilters(self, filters):
        # Filters arrive ANDed together; yield the ones Spark must still apply.
        for f in filters:
            if isinstance(f, EqualTo) and f.attribute == ("status",):
                self.pushed.append(f)          # handled at the source
            else:
                yield f

    def read(self, partition):
        import requests
        params = {f.attribute[0]: f.value for f in self.pushed}
        for row in requests.get(self.url, params=params).json():
            yield (row["id"], row["name"], row["status"])

Only conjunctions are pushed (a filter like a = 1 or b = 2 arrives as an empty list), and a reader can also yield pyarrow.RecordBatch objects instead of tuples, which the Spark docs say can improve performance by "up to one order of magnitude."

I haven't measured either path, so this is the one change on the list I'm ranking by what it removes rather than by what I saw: for a source backed by an API or a database, rows that are never fetched are the cheapest rows there are.

Skip pandas for grouped work

Per-group logic (normalize within each location, fit a model per customer) has long meant applyInPandas, which converts every group from Arrow into a pandas DataFrame and back. Spark 4.0 added applyInArrow (SPARK-40559), and each group now reaches your function as a pyarrow.Table, in the format it crossed the boundary in:

def normalize(table: pa.Table) -> pa.Table:
    totals = table.column("total")
    mean, std = pc.mean(totals).as_py(), pc.stddev(totals).as_py()
    normalized = (pa.array([0.0] * len(totals)) if not std
                  else pc.divide(pc.subtract(totals, mean), std))
    return pa.Table.from_arrays([table.column("order_id"), totals, normalized],
                                names=["order_id", "total", "normalized"])

result = orders.groupBy("location_id").applyInArrow(
    normalize, schema="order_id long, total double, normalized double")

This is the same trade as step 2, one level up: it removes a conversion, not a crossing, so the payoff should track what that conversion cost in my UDF numbers (little for numeric groups, more for string-heavy ones, more for larger groups). I haven't benchmarked the grouped APIs, so I won't put a number on it.

Take the JVM out of the client

Spark Connect is on this list for a different reason than the others. It doesn't make a query faster; it removes the JVM from every client process. Classic PySpark embeds a full Spark driver in each notebook, scheduled job, and CI container. With Connect, the client sends unresolved plans over gRPC and gets Arrow batches back, and the thin pyspark-client package has no JVM and needs no JDK. When I measured it for my own Connect demos on 4.2, the numbers were less dramatic than the pitch:

	`pyspark` (full)	`pyspark-client`
PyPI artifact	429.28 MB	1.60 MB
installed `pyspark/`	481 MB	19 MB
Spark JARs on the client	456 MB, 276 files	none
virtual env with Connect dependencies	820 MB	359 MB

The last row is the honest one: pyarrow, pandas, and numpy are required dependencies, so the environment shrinks by less than half.

What you actually get is a client with no JVM to start, no JDK to install, and no way for a crashed notebook to take the driver down with it. What you give up is anything that needs the JVM on your side: RDDs, SparkContext, custom accumulators. ML on Connect has been supported since 4.0 and generally available since 4.1, so that's no longer on the list of reasons to stay classic.

Give Python the memory it actually uses

The last change is the one that turns slow jobs into failed ones when it's wrong. Python workers live outside the JVM heap, so the settings most Spark tuning advice is about don't cover them:

The rule is that the container has to fit the heap, the overhead, and the OS, and that Python's share has to come out of the overhead. If you use toPandas() or pandas UDFs, spark.sql.execution.arrow.pyspark.selfDestruct.enabled frees Arrow buffers as they're converted, so the same data doesn't sit in memory twice.

Beyond that, the usual advice still holds: let Adaptive Query Execution size shuffle partitions at runtime (it's been on by default since 3.2, and 4.1 extended it to stateless streaming, SPARK-53941, which can change a streaming query's behaviour after an upgrade), and broadcast small dimension tables instead of shuffling them. None of that is specific to Python, which is why it's last on a list about Python.

What's coming in 4.3

The 4.3 branch hasn't been released (its first release candidate failed its vote on September 6, and there's no new date), but it shows where the boundary is heading, and it confirms the order above.

The most interesting change goes around the boundary instead of through it: experimental transpilation of Python UDFs into Spark expressions (SPIP SPARK-54783, spark.sql.experimental.optimizer.transpilePyUDFs). A simple enough Python function would never cross into Python at all, which is change number 1 done for you. It's off by default, requires ANSI mode, and only works in classic mode, not Connect, but it's the only approach on the branch that could get under the per-row ceiling rather than lowering the cost of hitting it.

Alongside it are smaller changes to the crossing itself: pipelined, asynchronous transfer of UDF batches (spark.python.udf.pipelined.enabled, SPARK-56642, off by default) and a cap on bytes per UDF batch (SPARK-57593). A proposal to make Spark better on a single machine also adds an Arrow format for Spark's in-memory cache (SPARK-57268).

Two things haven't moved. PySpark still doesn't support pandas 3, even on the 4.3 branch; it pins pandas below 3.0, and the ticket to support it (SPARK-55139) is open, so the PyArrow-backed string type pandas 3 made the default in January won't help pandas UDFs yet. And Real-Time Mode in 4.3 gains stateful operators, but still can't run Python UDFs (SPARK-57237 is open).

Housekeeping to plan for: 4.2 raised the minimum PyArrow to 18.0.0 and dropped PyPy, and 4.3 drops Python 3.10 (SPARK-56889).

I opened with a UDF that took 10 seconds at 100 million rows and was the fastest Python option at 100 thousand. The reason is the same one behind every change on this list: your function was never the expensive part.

The crossing was, and the best PySpark optimization is still the one that decides the data doesn't need to cross.

Stop Blaming Serialization for Slow Python UDFs

lnc — Wed, 14 Oct 2026 05:19:38 +0000

The benchmark that tested the wrong Arrow

Earlier this year I built a benchmark to answer a question that comes up constantly: if you need custom Python logic in Spark, which kind of UDF should you write, and what does it cost? It compared seven options across three workloads (trivial arithmetic, a string transform, and a normal CDF) at scales from 100K to 100M rows.

The result my write-up leaned on hardest was a pair of rows that came out almost identical: a plain @udf ran at 48.5x the cost of the equivalent built-in function at 50M rows, and the same @udf "with Arrow turned on" ran at 48.9x (69.2x against 69.0x at 100M). I called it the key insight: switching transport from pickle to Arrow buys nothing when your function still executes one row at a time.

The conclusion about per-row execution is right, but the evidence wasn't what I thought. This is the setting the benchmark toggled:

# What my benchmark flipped for its "@udf + Arrow" case:
spark.conf.set("spark.sql.execution.arrow.pyspark.enabled", "true")
# ...which only governs DataFrame.toPandas() and createDataFrame(pandas_df).
# It never touches Python UDFs.

# What actually moves a regular Python UDF onto Arrow:
spark.conf.set("spark.sql.execution.pythonUDF.arrow.enabled", "true")   # session-wide
@udf(DoubleType(), useArrow=True)                                        # or per UDF
def f(x): ...

The PySpark documentation is explicit that arrow.pyspark.enabled controls pandas conversions. So my two nearly identical rows were the same pickled UDF measured twice, and the gap between them (under 1%) is just how much two runs of the same thing vary.

I had built an A/A test by accident, in a benchmark that existed precisely because Spark's "Arrow" naming is easy to confuse.

That mistake is the best evidence I have for the actual thesis: the cost of a Python UDF is decided far more by what your function receives (a single value, a pandas Series, or an Arrow array) than by whether Arrow carries the bytes, and because "Arrow" names several different mechanisms, it's easy to flip the switch that sounds right and optimize nothing. Here is what each of them actually does:

What you set	What it controls	Affects your UDF?
`spark.sql.execution.arrow.pyspark.enabled`	`toPandas()` and `createDataFrame(pandas_df)` conversions	No
`@udf(useArrow=True)` or `spark.sql.execution.pythonUDF.arrow.enabled`	Moves a regular UDF to Arrow batches (`ArrowEvalPython`); your function still gets one scalar per call. On by default since Spark 4.2	Yes, 1.4x to 2.4x
`@pandas_udf`	Arrow batches; your function gets a `pandas.Series` per call	Yes, vectorized
`@arrow_udf` (Spark 4.1+, SPARK-53014)	Arrow batches; your function gets a `pyarrow.Array` per call	Yes, vectorized, no pandas

Three choices hiding behind every decorator

Every Python UDF call makes three independent choices: how data crosses from the JVM to the Python worker (transport), which physical operator Spark plans (BatchEvalPython or ArrowEvalPython), and what your function receives once the data arrives (the execution model). The last one determines how many times Python has to be entered.

UDF type	Operator	Transport	Calls into your Python	Receives
`@udf` (default, pre-4.2)	`BatchEvalPython`	pickle	once per row	scalar
`@udf(useArrow=True)`	`ArrowEvalPython`	Arrow	once per row	scalar
`@pandas_udf`	`ArrowEvalPython`	Arrow	once per batch	`pandas.Series`
`@arrow_udf`	`ArrowEvalPython`	Arrow	once per batch	`pyarrow.Array`

This is the exact code the benchmark ran for the arithmetic workload (x * 2 + 1):

@udf(DoubleType())                    # once per row, pickle
def py_arith(x):
    return float(x) * 2 + 1 if x is not None else None

@udf(DoubleType(), useArrow=True)     # once per row, Arrow batches
def use_arrow_arith(x):
    return float(x) * 2 + 1 if x is not None else None

@pandas_udf(DoubleType())             # once per batch, pandas.Series
def pandas_arith(x):
    return x * 2 + 1

@arrow_udf(DoubleType())              # once per batch, pyarrow.Array
def arrow_arith(x: pa.Array) -> pa.Array:
    return pc.add(pc.multiply(x, 2), 1)

And for the string workload (upper(x) + '_SUFFIX'), where the two vectorized options separate:

@pandas_udf(StringType())
def pandas_string(x):
    return x.str.upper() + "_SUFFIX"

@arrow_udf(StringType())
def arrow_string(x: pa.Array) -> pa.Array:
    upper = pc.utf8_upper(x)
    suffix = pa.array(["_SUFFIX"] * len(upper), type=pa.string())   # see note below
    return pc.binary_join_element_wise(upper, suffix, "")

What the measurements show

Median wall-clock seconds at 50M rows, with the multiple of the built-in time in parentheses (the mislabeled "Arrow config" row is omitted, since it was a second run of the default @udf):

	arithmetic	string	normal CDF
built-in functions	0.107	0.486	0.176
SQL UDF	0.095 (0.9x)	0.477 (1.0x)	n/a
`@arrow_udf`	0.725 (6.8x)	1.545 (3.2x)	0.762 (4.3x)
`@pandas_udf`	0.789 (7.4x)	2.512 (5.2x)	0.819 (4.7x)
`@udf(useArrow=True)`	2.228 (20.8x)	4.430 (9.1x)	2.942 (16.7x)
`@udf` (default, pickle)	5.199 (48.5x)	6.216 (12.8x)	6.183 (35.1x)

And at 100M rows:

	arithmetic	string	normal CDF
built-in functions	0.147	0.910	0.267
`@arrow_udf`	1.332	2.904	1.333
`@pandas_udf`	1.400	4.743	1.449
`@udf(useArrow=True)`	4.188	8.735	5.658
`@udf` (default, pickle)	10.139	12.212	12.060

Three things stand out, and each one follows from the execution model rather than the transport.

Arrow batching helps a row-at-a-time UDF, but only about 2x. useArrow=True took arithmetic from 5.20s to 2.23s and the CDF from 6.18s to 2.94s, because ArrowEvalPython moves data in efficient batches instead of pickling rows. It still calls your function once per value, though, and that per-call cost is a hard ceiling: across every workload and scale, the row-at-a-time UDFs topped out at roughly 8 to 10 million rows per second (50M rows in 5.2s; 100M in 10.1s), which is about the rate at which CPython can be entered and exited no matter what's computed inside.

The vectorized UDFs cross into Python once per batch instead, and they were about 3x faster again than useArrow=True on arithmetic.

Between the vectorized options, the gap is almost entirely strings.

`@arrow_udf` speedup over `@pandas_udf`	arithmetic	string	CDF
50M rows	1.09x	1.63x	1.07x
100M rows	1.05x	1.63x	1.09x

That split is what you'd expect from how the two libraries lay out data. A numeric pandas Series is a NumPy array underneath, so converting from Arrow is cheap and the arithmetic is the same vectorized arithmetic either way. Strings are different; pandas has traditionally stored them as an array of Python objects, each its own heap allocation, while Arrow keeps them in one contiguous buffer plus an array of offsets. Converting to pandas means building millions of Python objects, and every operation afterwards walks memory scattered across the heap.

If anything the benchmark understates the gap, because my @arrow_udf string function builds a Python list of suffixes the length of every batch before joining them, which is exactly the per-element Python work pyarrow.compute exists to avoid. The CDF case is close for the opposite reason: pyarrow.compute has no erf, so my "Arrow" CDF converts to NumPy and calls SciPy, just like the pandas version does.

Scale changes which choice matters. Arithmetic workload, median seconds (the @arrow_udf column is blank below 50M because it wasn't a separate case in those earlier runs):

rows	built-in	`@arrow_udf`	`@pandas_udf`	`useArrow=True`	`@udf` (pickle)
100K	0.073		0.181	0.212	0.161
1M	0.077		0.200	0.248	0.258
10M	0.074		0.299	0.568	1.092
50M	0.107	0.725	0.789	2.228	5.199
100M	0.147	1.332	1.400	4.188	10.139

At 100K rows every Python UDF was within 2x to 6x of the built-ins, and the plain pickled @udf was actually the fastest Python option on arithmetic, because fixed costs (starting Python workers, setting up batches) dominate everything at that size. The tiers only separate cleanly somewhere around 10M rows.

If your job processes a few hundred thousand rows, UDF type is unlikely to be what's making it slow.

Why I'd trust the seconds over the multipliers

There's a second correction to my own write-up, and it applies to almost any benchmark reported as "Nx slower." My report said row-at-a-time UDFs showed superlinear growth, from 48.5x at 50M to 69.2x at 100M. That's true of the ratio and false of the thing you care about:

default `@udf`, arithmetic	50M	100M	growth for 2x rows
`@udf` absolute time	5.199s	10.139s	1.95x (linear)
built-in absolute time	0.107s	0.147s	1.37x
ratio	48.5x	69.2x	"superlinear"

The ratio grew because its denominator grew slowly; the built-in at this scale is mostly fixed overhead.

The baseline has another problem I can't fully explain away. SQL UDFs, which compile to the same plan as built-ins, came out slightly faster than them at every scale (0.64x to 0.98x). The likeliest cause is ordering: the built-in case always ran first in each workload and may have absorbed start-up cost despite the warmup. If so, every multiplier here is somewhat inflated. An earlier set of my own 50M runs, before I raised driver memory, had the built-in at 0.25s and the pickled @udf at about 22x, while the @udf's own time moved only about 10%.

The headline multiplier more than doubled on the strength of the baseline alone.

What has changed, and what I didn't measure

Spark 4.2 turns Arrow optimization on by default for regular Python UDFs, after an attempt to do the same for 4.0 was reverted (SPARK-48516). On 4.2 a plain @udf should therefore behave like my useArrow=True rows, a free 2x for anyone who never touches the setting; I haven't rerun the benchmark on 4.2, so that's an expectation, not a measurement.

I also only benchmarked scalar UDFs. The partition-level APIs mapInPandas and mapInArrow, and the grouped applyInPandas and applyInArrow, follow the same logic about what your function receives (the Arrow variants skip pandas conversion entirely), and they're the right tools when you need to change the number of rows or see a whole partition or group at once. I have no measurements for them, and I'd rather say so than borrow numbers.

The benchmark's own limits:

One machine, local[*], via Spark Connect; hardware not recorded.
Synthetic, low-cardinality data (spark.range plus two derived columns, short strings), cached in memory before timing.
Each case materialized with a count, not written anywhere.
One warmup, median of three timed runs, no variance recorded.
The 100K to 10M runs predate the 32GB driver setting and the separate @arrow_udf case, so there's no genuine @arrow_udf data below 50M.

None of this changes the ordering, which held everywhere, but the exact multipliers are specific to this setup.

Choosing, based on what was measured

Whichever you choose, check which "Arrow" your setting actually refers to. Mine didn't, and I wrote a benchmark about exactly this.

DEV Community: lnc

Apache Iceberg v4 Column Updates Are About to Change Feature Engineering

Efficient Column Updates in a Nutshell

Current Column Updates in Iceberg Are Expensive

Byte by Byte

Anatomy of a Column File

Stitching It Back Together on Read

Explaining every row

How a column update effects manifest

But why not directly in Parquet?

State of the Implementation

What you should know and how to proceed

Further Reading

Spark Connect from Rust: Pull Spark outputs into DuckDB and Polars

Setting the Rust-Native Scene

What the Rust client actually is

Setting up the environment

Step 1: Some data to play with

Step 2: Getting a result out as Arrow (four ways, and one of them is wrong)

Step 3: DuckDB 2.0 gets the batch

Step 4: Polars gets Spark's aggregate

Step 5: Sending the answer back to Spark

Spark Type Mapping to DuckDB and Polars

Limitations

Summarizing the Stack

Further reading

The Ultimate Guide to Apache Spark 4.3

A release about time (mostly)

Stateful Real-time Mode

SCD Type 2 in Auto CDC

Project Feather: faster on small data

Spark Connect: operation IDs and REST endpoints

PySpark: DataFrame.zip and UDF transpilation

Spark SQL: ASOF JOIN, JSON_TABLE, VARIANT edits and nanoseconds

Reading files inside archives

Other changes in Spark 4.3

In Summary

Further reading

Building a Spark Connect Client in OCaml because I'm crazy and you're not

But Lisa, surely you must have a good reason for using OCaml?

The Roles and Responsibilities of a Connect Client

Step 1: Getting the protos to compile

Step 2: Speak gRPC

Step 3: A session, and the first RPC

Step 4: Build plans instead of running them

Step 5: Execute, and decode the Arrow that comes back

Step 6: Sending data to the server

Step 7: Errors, the way Spark sends them

Putting it together

Testing its Production Readiness

Every message, tested against three Spark releases

Case Study: A small options desk

What's left on my to do list, and where the official clients spend their time

Further reading

Hosting an Entire Open Lakehouse on Port 8080

Why on my own machine

Starting from the data

A catalog I had to build

Streams: Kafka and Zookeeper

Data that misbehaves

Spark, twice

Orchestration: Airflow

One CLI, built for agents

Credentials, everywhere

Testing it, and CI

Declarative pipelines and real time

Scheduling SDP: SparkPipelinesOperator

Converging streaming and batch

Building Iceberg for a Spark that doesn't exist

Why run Spark Connect at home at all

Experimental agents

Someone else's machine

Where it is now

All About Iceberg's File Format API: simplifying the engine/format interface and Parquet's multimodal future

What is the file format API?

Reducing Format Model x Reader Complexity

Into the weeds

The registry key

What happens on a read

How models get registered

Scheduling SDP: `SparkPipelinesOperator`