DEV Community: lnc

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.

Explaining the Python-JVM Boundary in Spark, and What Arrow Changed

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.

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.

The rest of this post is the story of Spark adopting it, one door at a time.

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

Before going further it's worth seeing what's moving, because every later boundary sends the same thing. Arrow's IPC stream format is 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.

The strings are the detail to hold on to: in Arrow, a million strings are one buffer and an array of offsets. 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 I'll leave numbers out of this section; 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))

Line the UDF types up by what happens to the data and the progression is plain:

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).

That's the strings detail from the socket section paying off. 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 there's no new 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.

How to Build an Open Lakehouse on Your Laptop

lnc — Thu, 01 Oct 2026 19:42:20 +0000

An open lakehouse is one where every layer (storage, table format, engine, catalog, and the ML and AI tools on top) is built on open standards, so no layer is locked to a single vendor. I wrote about what that means and why it matters in What is an open lakehouse? Open data standards, explained. *The practical payoff is that a team can change processing engines, add a query engine, adopt a second table format or swap a model framework without re-platforming the layers above or below. *

In practice, a each of these open source technologies will be a hosted service that each do one job: object storage holds the files, a table format turns files into tables, a catalog names the tables, Spark reads and writes them, a stream brings in what's happening now, a pipeline framework derives the tables people actually query, an orchestrator runs it all on a schedule, and experiment tracking records what you compute from it. This post builds all of it on your laptop, starting from an empty folder, with nothing but Docker and Python. Every file you need is in the post, in full: a compose.yaml that grows by a few services per section, a handful of config files, and short Python scripts. You run plain docker compose and python commands, and nothing else.

This post builds this stack manually layer by layer. If you want to just have a full packaged lakehouse, check out the project open-lakehouse. It comes with several agent skills, a cli, and a reference in case you get stuck on one of the steps.

What this tutorial covers

Layer	What runs	Image or package	Port
Storage	SeaweedFS, an S3-compatible object store	`chrislusf/seaweedfs:3.80`	8333
Compute	a Spark master, a worker and a Spark Connect server	`apache/spark:4.2.0`	15002 (UI 8080)
Table format	Delta Lake, with UniForm for Iceberg readers	Delta 4.4.0, loaded by Spark
Catalog	Unity Catalog OSS	`unitycatalog/unitycatalog:v0.6.0`	8081
Iceberg readers	PyIceberg and DuckDB	on your machine
Pipelines	Spark Declarative Pipelines	part of Spark 4.2
Event log	Kafka, single node, KRaft mode	`apache/kafka:4.2.0`	9092
Streaming	Structured Streaming, including Real-Time Mode	part of Spark 4.2
Orchestration	Airflow, in standalone mode	`apache/airflow:3.3.2-python3.12`	8085
ML	MLflow tracking server	`ghcr.io/mlflow/mlflow:v3.16.1`	5000

Everything runs on one Spark version, 4.2.0, and the pieces that plug into Spark are pinned to match it. There's no separate database: Airflow and MLflow keep their metadata in SQLite and Unity Catalog in its embedded H2 database, which is plenty for one laptop.

Basic Requirements

Docker with the Compose plugin, Compose v2.17 or newer. On Linux that's Docker Engine plus the docker-compose-plugin package; on macOS, Docker Desktop; on Windows, Docker Desktop with WSL2, and you follow the Linux steps inside your WSL distribution (keep the project folder in your WSL home directory, not under /mnt/c).
Python 3.10 or newer, with venv (on Ubuntu and WSL, sudo apt install python3-venv if python3 -m venv complains).
Memory and disk. The whole stack uses about 9 GB of RAM once everything is up, so a 16 GB machine is comfortable; in Docker Desktop, raise the memory limit under Settings > Resources to at least 10 GB. Plan for about 12 GB of disk for images and downloads.

I ran every step below on Ubuntu 24.04 with Docker Engine, copying each file and command out of this post into an empty folder.

Every port the stack publishes is written in compose.yaml as ${NAME:-default}, for example "${S3_PORT:-8333}:8333". If one of the defaults is already taken on your machine, pick another in a .env file next to compose.yaml (a line like S3_PORT=18333; Compose reads that file on every command), and point the scripts at it with the matching variable from common.py below (export S3_ENDPOINT=http://localhost:18333).

A folder and a Python environment

Make an empty folder and give it a virtual environment with the clients this post uses:

mkdir open-lakehouse && cd open-lakehouse
python3 -m venv .venv
source .venv/bin/activate
pip install pyspark-client==4.2.0 pandas==2.3.3 boto3==1.43.93 kafka-python==2.3.0 pyiceberg==0.12.0 duckdb==1.5.6 mlflow-skinny==3.16.1

pyspark-client is PySpark without the JVM: a thin client that sends query plans to a Spark Connect server and gets results back, so your machine never runs Spark itself. The others are the S3 client (boto3), a Kafka producer for the test data, the two Iceberg readers, and the MLflow client. (pandas is pinned below 3.0, which PySpark 4.2 doesn't fully support yet.) Open a new terminal later and you'll need source .venv/bin/activate again in that folder.

Where everything goes

Everything in this post lives in that one open-lakehouse folder. By the end it looks like this:

open-lakehouse/
├── .venv/                      the Python environment you just made
├── compose.yaml                every service, added layer by layer
├── s3.json                     SeaweedFS access keys
├── server.properties           Unity Catalog settings
├── spark-conf/
│   └── spark-defaults.conf     Spark settings, mounted into the Spark containers
├── pipeline/
│   ├── spark-pipeline.yml      the declarative pipeline's spec
│   └── transformations/
│       ├── bronze.py
│       ├── silver.sql
│       └── gold.sql
├── dags/
│   ├── medallion.py            the Airflow DAG that runs the pipeline
│   └── maintenance.py          the Airflow DAG that compacts and vacuums
├── common.py                   shared endpoints, imported by every script below
├── create_bucket.py
├── hello_spark.py
├── generate_orders.py
├── load_orders.py
├── register_orders.py
├── enable_uniform.py
├── read_iceberg.py
├── prepare_medallion.py
├── read_gold.py
├── stream_clean.py
├── stream_totals.py
├── check_stream.py
└── log_run.py

The label above each code block is the file's path, so File: open-lakehouse/spark-conf/spark-defaults.conf means create that file in that subfolder. The Python scripts all sit at the top of the folder, next to common.py, so they can import it. Run every command from inside open-lakehouse, with the virtual environment active (source .venv/bin/activate); docker compose finds compose.yaml there, and python generate_orders.py finds the script there.

Every script imports its endpoints and clients from one small module, so the addresses live in one place:

File: open-lakehouse/common.py

"""Endpoints and clients that every script in this tutorial shares."""

import json
import os
import urllib.request

import boto3
from pyspark.sql import SparkSession

SPARK_REMOTE = os.environ.get("SPARK_REMOTE", "sc://localhost:15002")
S3_ENDPOINT = os.environ.get("S3_ENDPOINT", "http://localhost:8333")
UC_URL = os.environ.get("UC_URL", "http://localhost:8081")
KAFKA_BOOTSTRAP = os.environ.get("KAFKA_BOOTSTRAP", "localhost:9092")
MLFLOW_TRACKING_URI = os.environ.get("MLFLOW_TRACKING_URI", "http://localhost:5000")
S3_KEY, S3_SECRET = "lakehouse", "lakehouse-secret"


def spark():
    """A Spark Connect session: a thin client, the work runs in the cluster."""
    return SparkSession.builder.remote(SPARK_REMOTE).getOrCreate()


def s3():
    """An S3 client for SeaweedFS."""
    return boto3.client(
        "s3",
        endpoint_url=S3_ENDPOINT,
        aws_access_key_id=S3_KEY,
        aws_secret_access_key=S3_SECRET,
        region_name="us-east-1",
    )


def uc(method, path, body=None):
    """Call the Unity Catalog REST API and return the JSON reply."""
    request = urllib.request.Request(
        f"{UC_URL}/api/2.1/unity-catalog{path}",
        data=json.dumps(body).encode() if body else None,
        headers={"Content-Type": "application/json"},
        method=method,
    )
    with urllib.request.urlopen(request) as reply:
        return json.load(reply)

The defaults are the ports the stack publishes on your machine. s3() talks to the object store, uc() to the catalog's REST API, and spark() opens a Spark Connect session, which is all the scripts need to run on the cluster.

Storage: SeaweedFS

A lakehouse keeps its tables as ordinary files in object storage: Parquet data files plus the table format's own log next to them. SeaweedFS is a small open-source object store that speaks the S3 API, so Spark writes to it the same way it would write to Amazon S3, and every tool in this post that reads S3 can read it too.

SeaweedFS needs to know who may use its S3 API. This identities file creates one user with a local key pair and full rights:

File: open-lakehouse/s3.json

{
  "identities": [
    {
      "name": "lakehouse",
      "credentials": [{ "accessKey": "lakehouse", "secretKey": "lakehouse-secret" }],
      "actions": ["Admin", "Read", "Write", "List", "Tagging"]
    }
  ]
}

Then the first version of compose.yaml, with one service:

File: open-lakehouse/compose.yaml

services:
  seaweedfs:
    image: chrislusf/seaweedfs:3.80
    entrypoint: weed
    command: server -dir=/data -s3 -s3.config=/etc/seaweedfs/s3.json -volume.max=0 -master.volumeSizeLimitMB=1024
    ports:
      - "${S3_PORT:-8333}:8333"
    volumes:
      - ./s3.json:/etc/seaweedfs/s3.json:ro
      - seaweedfs-data:/data
    healthcheck:
      test: wget -qO /dev/null http://127.0.0.1:8333/status
      interval: 5s
      retries: 30

volumes:
  seaweedfs-data:

weed server -s3 runs the whole object store in one process (the master, a volume server, the filer and the S3 gateway); -volume.max=0 and -master.volumeSizeLimitMB=1024 let it add 1 GB storage volumes as the data grows. The named volume seaweedfs-data keeps the files across restarts. The healthcheck asks the S3 gateway for its status, and --wait makes docker compose up return only once every service with a healthcheck reports healthy:

docker compose up -d --wait

To check it, create the bucket the rest of the stack uses, write an object, and list it back:

File: open-lakehouse/create_bucket.py

"""Create the `lakehouse` bucket, write one object and list it back."""

from common import s3

client = s3()
existing = [b["Name"] for b in client.list_buckets()["Buckets"]]
if "lakehouse" not in existing:
    client.create_bucket(Bucket="lakehouse")

client.put_object(Bucket="lakehouse", Key="hello.txt", Body=b"hello, lakehouse\n")
for obj in client.list_objects_v2(Bucket="lakehouse")["Contents"]:
    print(obj["Key"], obj["Size"], "bytes")

python create_bucket.py

hello.txt 17 bytes

flowchart LR
  classDef new fill:#d97706,fill-opacity:0.72,stroke:#d97706,color:#ffffff
  classDef old fill:#4b5563,fill-opacity:0.72,stroke:#4b5563,color:#ffffff
  st_PY["your Python"]:::new
  subgraph st_ST["storage"]
    st_SW[("SeaweedFS :8333<br/>bucket lakehouse")]:::new
  end
  st_PY -->|"S3 API"| st_SW

Compute: Spark 4.2 and Spark Connect

Spark is the engine: it reads and writes every table, and everything above this layer either feeds it or tells it what to run. The stack runs a small standalone cluster (a master and one worker) plus a Spark Connect server. Connect splits the classic Spark application in two: the server owns the driver, the catalog settings and the storage credentials, and clients send it query plans over gRPC as thin Python processes. Every client in this post, from your terminal to Airflow, talks to the same server.

Spark reads its settings from spark-defaults.conf. Make a folder for it:

mkdir spark-conf

File: open-lakehouse/spark-conf/spark-defaults.conf

# Delta Lake
spark.sql.extensions                             io.delta.sql.DeltaSparkSessionExtension
spark.sql.catalog.spark_catalog                  org.apache.spark.sql.delta.catalog.DeltaCatalog

# SeaweedFS, through Hadoop's S3A connector, for both s3:// and s3a:// paths
spark.hadoop.fs.s3a.endpoint                     http://seaweedfs:8333
spark.hadoop.fs.s3a.access.key                   lakehouse
spark.hadoop.fs.s3a.secret.key                   lakehouse-secret
spark.hadoop.fs.s3a.path.style.access            true
spark.hadoop.fs.s3a.connection.ssl.enabled       false
spark.hadoop.fs.s3.impl                          org.apache.hadoop.fs.s3a.S3AFileSystem
spark.hadoop.fs.AbstractFileSystem.s3.impl       org.apache.hadoop.fs.s3a.S3A

# Resources: a cap, so other applications still get cores, and idle executors go back
spark.driver.memory                              2g
spark.executor.memory                            1g
spark.executor.cores                             2
spark.cores.max                                  4
spark.dynamicAllocation.enabled                  true
spark.dynamicAllocation.shuffleTracking.enabled  true
spark.dynamicAllocation.executorIdleTimeout      60s
spark.dynamicAllocation.shuffleTracking.timeout  120s
spark.dynamicAllocation.cachedExecutorIdleTimeout 300s

# Where --packages keeps the JARs it downloads (a volume, so it happens once)
spark.jars.ivy                                   /opt/spark/work-dir/ivy

The first block turns on Delta Lake. The second points Hadoop's S3A connector at SeaweedFS for both s3a:// and s3:// paths, so you can write s3://lakehouse/... everywhere. The third keeps the Connect server, a single long-running application, from holding the whole worker forever: it caps the server at four cores, so a job you submit to the cluster later still gets some, and hands idle executors back to the worker after a minute or two (shuffle tracking alone keeps an executor as long as it holds shuffle files, which in a long-lived Connect session can mean forever). The last line tells Spark where to cache the JARs it downloads.

Now add the three Spark services to compose.yaml, under services::

File: open-lakehouse/compose.yaml, add under services:

  spark-master:
    image: apache/spark:4.2.0
    hostname: spark-master
    command: /opt/spark/bin/spark-class org.apache.spark.deploy.master.Master --host spark-master
    ports:
      - "${SPARK_UI_PORT:-8080}:8080"
    healthcheck:
      test: curl -sf http://localhost:8080 > /dev/null
      interval: 5s
      retries: 30

  spark-worker:
    image: apache/spark:4.2.0
    hostname: spark-worker
    command: /opt/spark/bin/spark-class org.apache.spark.deploy.worker.Worker --host spark-worker spark://spark-master:7077
    environment:
      SPARK_WORKER_CORES: "6"
      SPARK_WORKER_MEMORY: 4g
    volumes:
      - spark-work:/opt/spark/work-dir
    depends_on:
      spark-master:
        condition: service_healthy

  spark-connect:
    image: apache/spark:4.2.0
    hostname: spark-connect
    command: >
      /opt/spark/sbin/start-connect-server.sh
      --master spark://spark-master:7077
      --packages io.delta:delta-spark_4.2_2.13:4.4.0,io.delta:delta-iceberg_2.13:4.4.0,io.unitycatalog:unitycatalog-spark_4.2_2.13:0.6.0,org.apache.hadoop:hadoop-aws:3.5.0,org.apache.spark:spark-sql-kafka-0-10_2.13:4.2.0
      --exclude-packages io.delta:delta-spark_4.1_2.13
    environment:
      SPARK_NO_DAEMONIZE: "1"
    ports:
      - "${SPARK_CONNECT_PORT:-15002}:15002"
    volumes:
      - ./spark-conf:/opt/spark/conf
      - spark-work:/opt/spark/work-dir
    depends_on:
      spark-master:
        condition: service_healthy
    healthcheck:
      test: ["CMD", "bash", "-c", "echo > /dev/tcp/localhost/15002"]
      interval: 5s
      start_period: 15m

and their shared volume, under volumes: at the bottom:

File: open-lakehouse/compose.yaml, add under volumes:

  spark-work:

All three run the official apache/spark:4.2.0 image, with no custom build. The master and worker are Spark's standalone cluster; the worker offers six cores and 4 GB to applications. The Connect server is started with --packages, which resolves Spark's plugins from Maven Central when it starts and puts them on the classpath of the driver and every executor:

Package	Version	What it adds
`io.delta:delta-spark_4.2_2.13`	4.4.0	Delta Lake, built for Spark 4.2
`io.delta:delta-iceberg_2.13`	4.4.0	UniForm: Delta tables that also write Iceberg metadata
`io.unitycatalog:unitycatalog-spark_4.2_2.13`	0.6.0	Spark's connector to Unity Catalog, matching the server
`org.apache.hadoop:hadoop-aws`	3.5.0	the S3A filesystem; it must match the Hadoop inside the Spark image, and it brings the AWS SDK with it
`org.apache.spark:spark-sql-kafka-0-10_2.13`	4.2.0	the Kafka source and sink for Structured Streaming

--exclude-packages drops one dependency: delta-iceberg asks for the Spark 4.1 build of Delta, and the Spark 4.2 build is already on the list. The first start downloads about 800 MB (most of it the AWS SDK), so it takes longer than the later ones, which reuse the downloads the spark-work volume keeps (docker compose logs -f spark-connect in another terminal shows the progress). The worker mounts the same volume, which the streaming section uses for its checkpoints.

Here's the whole file at this point:

File: open-lakehouse/compose.yaml

services:
  seaweedfs:
    image: chrislusf/seaweedfs:3.80
    entrypoint: weed
    command: server -dir=/data -s3 -s3.config=/etc/seaweedfs/s3.json -volume.max=0 -master.volumeSizeLimitMB=1024
    ports:
      - "${S3_PORT:-8333}:8333"
    volumes:
      - ./s3.json:/etc/seaweedfs/s3.json:ro
      - seaweedfs-data:/data
    healthcheck:
      test: wget -qO /dev/null http://127.0.0.1:8333/status
      interval: 5s
      retries: 30

  spark-master:
    image: apache/spark:4.2.0
    hostname: spark-master
    command: /opt/spark/bin/spark-class org.apache.spark.deploy.master.Master --host spark-master
    ports:
      - "${SPARK_UI_PORT:-8080}:8080"
    healthcheck:
      test: curl -sf http://localhost:8080 > /dev/null
      interval: 5s
      retries: 30

  spark-worker:
    image: apache/spark:4.2.0
    hostname: spark-worker
    command: /opt/spark/bin/spark-class org.apache.spark.deploy.worker.Worker --host spark-worker spark://spark-master:7077
    environment:
      SPARK_WORKER_CORES: "6"
      SPARK_WORKER_MEMORY: 4g
    volumes:
      - spark-work:/opt/spark/work-dir
    depends_on:
      spark-master:
        condition: service_healthy

  spark-connect:
    image: apache/spark:4.2.0
    hostname: spark-connect
    command: >
      /opt/spark/sbin/start-connect-server.sh
      --master spark://spark-master:7077
      --packages io.delta:delta-spark_4.2_2.13:4.4.0,io.delta:delta-iceberg_2.13:4.4.0,io.unitycatalog:unitycatalog-spark_4.2_2.13:0.6.0,org.apache.hadoop:hadoop-aws:3.5.0,org.apache.spark:spark-sql-kafka-0-10_2.13:4.2.0
      --exclude-packages io.delta:delta-spark_4.1_2.13
    environment:
      SPARK_NO_DAEMONIZE: "1"
    ports:
      - "${SPARK_CONNECT_PORT:-15002}:15002"
    volumes:
      - ./spark-conf:/opt/spark/conf
      - spark-work:/opt/spark/work-dir
    depends_on:
      spark-master:
        condition: service_healthy
    healthcheck:
      test: ["CMD", "bash", "-c", "echo > /dev/tcp/localhost/15002"]
      interval: 5s
      start_period: 15m

volumes:
  seaweedfs-data:
  spark-work:

Start the cluster. The Connect server's healthcheck passes once its gRPC port accepts connections, so --wait returns when the server is ready for queries:

docker compose up -d --wait

Then send it a query:

File: open-lakehouse/hello_spark.py

"""Send one query to the cluster through Spark Connect."""

from common import SPARK_REMOTE, spark

session = spark()
print(f"Connected to Spark {session.version} at {SPARK_REMOTE}")
session.range(1_000_000).selectExpr("sum(id) AS total").show()

python hello_spark.py

Connected to Spark 4.2.0 at sc://localhost:15002
+------------+
|       total|
+------------+
|499999500000|
+------------+

The sum ran on the worker; your terminal only sent the plan and printed the result. The Spark UI at http://localhost:8080 lists the worker and the running "Spark Connect server" application.

flowchart LR
  classDef new fill:#d97706,fill-opacity:0.72,stroke:#d97706,color:#ffffff
  classDef old fill:#4b5563,fill-opacity:0.72,stroke:#4b5563,color:#ffffff
  sp_PY["PySpark client"]:::new
  sp_SC["Spark Connect<br/>:15002"]:::new
  sp_CL["master + worker"]:::new
  sp_SW[("SeaweedFS")]:::old
  sp_PY -->|"gRPC"| sp_SC
  sp_SC --> sp_CL
  sp_CL -->|"S3A"| sp_SW

Some data: a ghost-kitchen generator

A lakehouse needs something to hold. This generator makes up orders for a ghost kitchen, a delivery-only restaurant cooking for several brands in four cities. Each order produces three events (order_created, order_ready, delivered), with the brand, item count and total as a JSON string in body. It misbehaves on purpose, the way real feeds do: some events arrive twice and a few lose their location, so the later layers have something to clean up. It writes a week of history to the bucket as Parquet, or streams live events into Kafka for the streaming section:

File: open-lakehouse/generate_orders.py

"""Ghost-kitchen orders: a week of history for the bucket, or a live stream for Kafka.

    python generate_orders.py files    # 7 days of events -> s3://lakehouse/landing/
    python generate_orders.py stream   # one day of events, live -> Kafka topic `orders`
"""

import io
import json
import random
import sys
import time
from datetime import datetime, timedelta, timezone

import pandas as pd
from common import KAFKA_BOOTSTRAP, s3

LOCATIONS = {1: "Oakland", 2: "Berkeley", 3: "San Francisco", 4: "San Jose"}
BRANDS = {"Pizza Planet": 16.0, "Wok This Way": 21.0, "Taco Loco": 12.5,
          "Curry House": 23.0, "Sushi Express": 29.0}
rng = random.Random(7)


def order(ts):
    """One order's events: created, ready in the kitchen, delivered."""
    order_id = f"{rng.getrandbits(40):010x}"
    location = rng.choice(list(LOCATIONS))
    brand, items = rng.choice(list(BRANDS)), rng.randint(1, 4)
    total = round(BRANDS[brand] * items * rng.uniform(0.9, 1.2), 2)
    steps = [("order_created", 0), ("order_ready", rng.randint(8, 20)),
             ("delivered", rng.randint(25, 50))]
    return [{
        "event_id": f"{order_id}-{n}",
        "event_type": kind,
        "ts": (ts + timedelta(minutes=minutes)).isoformat(timespec="seconds"),
        "location_id": location,
        "order_id": order_id,
        "body": json.dumps({"brand": brand, "items": items, "total": total}),
    } for n, (kind, minutes) in enumerate(steps)]


def messy(events):
    """Real feeds misbehave: a few events lose their location, some arrive twice."""
    out = []
    for event in events:
        if rng.random() < 0.01:
            event["location_id"] = None
        out.append(event)
        if rng.random() < 0.03:
            out.append(dict(event))
    return out


def week():
    start = datetime(2026, 9, 1)
    for day in range(7):
        events = []
        for _ in range(rng.randint(1200, 1800)):
            events += order(start + timedelta(days=day, seconds=rng.randint(36000, 79200)))
        yield (start + timedelta(days=day)).date(), messy(events)


def put_parquet(client, key, frame):
    buffer = io.BytesIO()
    frame.to_parquet(buffer, index=False)
    client.put_object(Bucket="lakehouse", Key=key, Body=buffer.getvalue())
    print(f"{key}  {len(frame):,} rows")


def write_files():
    client = s3()
    for day, events in week():
        frame = pd.DataFrame(events).astype({"location_id": "Int64"})  # keep the gaps as nulls
        put_parquet(client, f"landing/orders/{day}.parquet", frame)
    cities = pd.DataFrame({"location_id": list(LOCATIONS), "city": list(LOCATIONS.values())})
    put_parquet(client, "landing/locations/locations.parquet", cities)


def stream(per_second=50):
    from kafka import KafkaProducer

    producer = KafkaProducer(bootstrap_servers=KAFKA_BOOTSTRAP, key_serializer=str.encode,
                             value_serializer=lambda v: json.dumps(v).encode())
    rng.seed()  # new order ids on every run
    _, events = next(week())
    events.sort(key=lambda e: e["ts"])
    stamps = {}
    for n, event in enumerate(events, 1):
        if event["event_id"] not in stamps:  # a resent event keeps its first timestamp
            late = rng.random() < 0.02  # and a few arrive three minutes late
            now = datetime.now(timezone.utc) - timedelta(minutes=3 if late else 0)
            stamps[event["event_id"]] = now.isoformat(timespec="milliseconds")
        producer.send("orders", key=event["order_id"],
                      value={**event, "ts": stamps[event["event_id"]]})
        if n % 1000 == 0:
            print(f"sent {n:,} events")
        time.sleep(1 / per_second)
    producer.flush()
    print(f"sent {len(events):,} events to the topic orders")


if __name__ == "__main__":
    {"files": write_files, "stream": stream}[sys.argv[1]]()

python generate_orders.py files

landing/orders/2026-09-01.parquet  4,720 rows
landing/orders/2026-09-02.parquet  4,382 rows
landing/orders/2026-09-03.parquet  5,479 rows
landing/orders/2026-09-04.parquet  4,508 rows
landing/orders/2026-09-05.parquet  5,194 rows
landing/orders/2026-09-06.parquet  4,962 rows
landing/orders/2026-09-07.parquet  4,776 rows
landing/locations/locations.parquet  4 rows

It's seeded, so you get the same week I did. The files land in a landing/ prefix in the bucket, the usual first stop for raw data before it becomes a table.

Table format: Delta Lake

Parquet files in a bucket aren't a table yet. A table format adds a transaction log next to the data files, where each write is a new, atomic commit that says which files make up the table now. That log is what gives you ACID writes on object storage, readers that never see half a write, and time travel. This stack uses Delta Lake, the format Unity Catalog OSS writes.

This script writes the week's events as a Delta table in two commits, the first day and then the other six, then reads the table as it is now and as it was at version 0:

File: open-lakehouse/load_orders.py

"""Write the week's events as a Delta table in two commits, then time-travel."""

from common import s3, spark
from pyspark.sql import functions as f

TABLE_PATH = "s3://lakehouse/tables/orders_raw"

session = spark()
events = session.read.parquet("s3://lakehouse/landing/orders/").withColumn(
    "event_date", f.to_date("ts")
)

# Commit 0: the first day. Commit 1: the other six.
events.where("event_date = '2026-09-01'").write.format("delta").save(TABLE_PATH)
events.where("event_date > '2026-09-01'").write.format("delta").mode("append").save(TABLE_PATH)

session.sql(f"DESCRIBE HISTORY delta.`{TABLE_PATH}`").select(
    "version", "operation", "operationMetrics.numOutputRows"
).show()

now = session.read.format("delta").load(TABLE_PATH).count()
then = session.read.format("delta").option("versionAsOf", 0).load(TABLE_PATH).count()
print(f"rows now: {now:,}   rows at version 0: {then:,}\n")

# Underneath, it's files in the bucket: Parquet data plus the _delta_log.
for obj in s3().list_objects_v2(Bucket="lakehouse", Prefix="tables/orders_raw/")["Contents"]:
    print(obj["Key"])

python load_orders.py

+-------+---------+-------------+
|version|operation|numOutputRows|
+-------+---------+-------------+
|      1|    WRITE|        29301|
|      0|    WRITE|         4720|
+-------+---------+-------------+

rows now: 34,021   rows at version 0: 4,720

tables/orders_raw/_delta_log/00000000000000000000.crc
tables/orders_raw/_delta_log/00000000000000000000.json
tables/orders_raw/_delta_log/00000000000000000001.crc
tables/orders_raw/_delta_log/00000000000000000001.json
tables/orders_raw/_delta_log/_staged_commits/
tables/orders_raw/part-00000-7f3f327f-b1e1-4f75-99de-ccfbc82c1b65-c000.snappy.parquet
tables/orders_raw/part-00000-92a94a61-efd3-4a85-924d-e3481bde0ed6-c000.snappy.parquet
tables/orders_raw/part-00001-9db8f5d3-76b4-4a7b-972b-b3b51c7f29e3-c000.snappy.parquet
tables/orders_raw/part-00001-fd4c4fe3-30ec-43f1-a3b6-3b5128241edd-c000.snappy.parquet

Version 0 still exists, because the append added files and a commit rather than rewriting anything. Underneath, the table is Parquet files plus a _delta_log folder with one JSON commit per version (the .crc files are checksums, and _staged_commits/ is an empty folder Delta keeps for commits that go through a catalog). The log is the table, and this listing shows it: there are four Parquet files, but the commits list only three of them. The fourth is an empty file from a task that had no rows for the first day, and since no commit references it, no reader ever sees it.

flowchart LR
  classDef new fill:#d97706,fill-opacity:0.72,stroke:#d97706,color:#ffffff
  classDef old fill:#4b5563,fill-opacity:0.72,stroke:#4b5563,color:#ffffff
  dl_SC["Spark Connect"]:::old
  subgraph dl_SW["SeaweedFS"]
    dl_LOG["_delta_log<br/>a commit per version"]:::new
    dl_PQ["Parquet<br/>data files"]:::new
  end
  dl_SC -->|"commit"| dl_LOG
  dl_SC -->|"write"| dl_PQ
  dl_LOG -.->|"lists"| dl_PQ

Catalog: Unity Catalog OSS

A path like s3://lakehouse/tables/orders_raw works, but nobody wants to pass paths around. A catalog gives tables names (catalog.schema.table), records where each one lives, and is the one place engines ask about tables. Unity Catalog OSS also governs access and hands engines short-lived storage credentials for the tables they ask for, which is why it needs the S3 settings too:

File: open-lakehouse/server.properties

server.env=dev
server.authorization=disable

# The bucket Unity Catalog hands out credentials for, and the key pair it uses
s3.bucketPath.0=s3://lakehouse
s3.region.0=us-east-1
s3.accessKey.0=lakehouse
s3.secretKey.0=lakehouse-secret
s3.sessionToken.0=unused
s3.endpoint.0=http://seaweedfs:8333

s3.bucketPath.0 is the bucket root the credentials cover. UC 0.6.0 skips a bucket's settings unless all four credential fields are filled in, so the session token gets a placeholder; SeaweedFS ignores it.

Add the catalog server, under services::

File: open-lakehouse/compose.yaml, add under services:

  unity-catalog:
    image: unitycatalog/unitycatalog:v0.6.0
    ports:
      - "${UC_PORT:-8081}:8080"
    volumes:
      - ./server.properties:/home/unitycatalog/etc/conf/server.properties:ro
      - uc-data:/home/unitycatalog/etc/db
    healthcheck:
      test: wget -qO /dev/null http://127.0.0.1:8080/api/2.1/unity-catalog/catalogs
      interval: 5s
      retries: 30

and its volume, under volumes::

File: open-lakehouse/compose.yaml, add under volumes:

  uc-data:

Spark needs to know about the catalog too. Add these lines to the end of spark-conf/spark-defaults.conf:

File: open-lakehouse/spark-conf/spark-defaults.conf, add at the end

# Unity Catalog: Spark's catalog `lakehouse` is the Unity Catalog catalog `lakehouse`
spark.sql.catalog.lakehouse                      io.unitycatalog.spark.UCSingleCatalog
spark.sql.catalog.lakehouse.uri                  http://unity-catalog:8080
spark.sql.catalog.lakehouse.token                unused

A Unity Catalog connector serves one catalog, and the Spark catalog's name is the Unity Catalog name it serves, so lakehouse.tutorial.orders_raw in Spark is the table orders_raw in schema tutorial of the catalog lakehouse. Start Unity Catalog, then recreate the Connect server so it reads the new settings (its downloads are in the volume, so this takes seconds):

docker compose up -d --wait
docker compose up -d --wait --force-recreate spark-connect

This script creates the lakehouse catalog through the REST API (a catalog can't be created from Spark), registers the Delta table you wrote, queries it by name, and asks the catalog what it holds:

File: open-lakehouse/register_orders.py

"""Create the `lakehouse` catalog, register orders_raw in it, and query it by name."""

from common import spark, uc

if "lakehouse" not in [c["name"] for c in uc("GET", "/catalogs")["catalogs"]]:
    uc("POST", "/catalogs", {"name": "lakehouse", "storage_root": "s3://lakehouse/managed"})

session = spark()
session.sql("CREATE SCHEMA IF NOT EXISTS lakehouse.tutorial")
session.sql("""
    CREATE TABLE IF NOT EXISTS lakehouse.tutorial.orders_raw
    USING delta LOCATION 's3://lakehouse/tables/orders_raw'
""")
session.sql("""
    SELECT event_date, count(*) AS events
    FROM lakehouse.tutorial.orders_raw
    GROUP BY event_date ORDER BY event_date
""").show()

# Ask the catalog itself, over its REST API.
for table in uc("GET", "/tables?catalog_name=lakehouse&schema_name=tutorial")["tables"]:
    print(table["name"], table["table_type"], table["data_source_format"], table["storage_location"])

python register_orders.py

+----------+------+
|event_date|events|
+----------+------+
|2026-09-01|  4720|
|2026-09-02|  4382|
|2026-09-03|  5479|
|2026-09-04|  4508|
|2026-09-05|  5194|
|2026-09-06|  4962|
|2026-09-07|  4776|
+----------+------+

orders_raw EXTERNAL DELTA s3://lakehouse/tables/orders_raw

Two kinds of Delta table live in this catalog. orders_raw is external: you chose its location and registered it. The pipeline section creates catalog-managed tables, where the catalog picks the location under the catalog's storage_root (s3://lakehouse/managed here) and owns the table's lifecycle. The locations are s3:// URIs because that's how Unity Catalog records them; spark-defaults.conf maps s3:// to the S3A filesystem, so Spark follows them.

flowchart LR
  classDef new fill:#d97706,fill-opacity:0.72,stroke:#d97706,color:#ffffff
  classDef old fill:#4b5563,fill-opacity:0.72,stroke:#4b5563,color:#ffffff
  uc_SC["Spark Connect"]:::old
  uc_UC["Unity Catalog<br/>:8081"]:::new
  uc_SW[("SeaweedFS")]:::old
  uc_SC -->|"UC connector"| uc_UC
  uc_UC -.->|"locations,<br/>credentials"| uc_SW
  uc_SC -->|"S3A"| uc_SW

Open formats: read the same table as Iceberg

Delta is one open table format; Apache Iceberg is another, and plenty of engines read Iceberg. You don't have to pick one. Both keep the data in Parquet and differ in the metadata they put next to it, so one copy of the files can carry both kinds of metadata. Delta calls that UniForm: the table stays a Delta table that Spark writes, and with each commit Delta also writes Iceberg metadata for the same files. Delta writes, Iceberg engines read.

The conversion lives in the delta-iceberg package the Connect server already loaded. Three table properties turn it on: Iceberg as a universal format, the IcebergCompatV2 table feature, and column mapping by name, since Iceberg tracks columns by ID rather than by position:

File: open-lakehouse/enable_uniform.py

"""Turn on UniForm: orders_raw stays a Delta table and also gets Iceberg metadata."""

from common import s3, spark

spark().sql("""
    ALTER TABLE lakehouse.tutorial.orders_raw SET TBLPROPERTIES (
        'delta.universalFormat.enabledFormats' = 'iceberg',
        'delta.enableIcebergCompatV2' = 'true',
        'delta.columnMapping.mode' = 'name'
    )
""")

for obj in s3().list_objects_v2(Bucket="lakehouse", Prefix="tables/orders_raw/metadata/")["Contents"]:
    print(obj["Key"])

python enable_uniform.py

tables/orders_raw/metadata/00000-01bfacd6-e573-475c-917f-34bc3a647658.metadata.json
tables/orders_raw/metadata/ec2026c7-0ade-4f27-a333-598ff9fa2096-m0.avro
tables/orders_raw/metadata/snap-2861189516262794144-1-ec2026c7-0ade-4f27-a333-598ff9fa2096.avro

The ALTER was itself a Delta commit, and UniForm converted it: a metadata.json, a manifest list (snap-...) and a manifest (...-m0), all pointing at the Parquet files that were already there. Nothing was rewritten; Iceberg readers match the existing files' columns by name.

Now read the table as Iceberg, with two readers that know nothing about Delta and don't use Spark: PyIceberg, Iceberg's Python library, and DuckDB with its iceberg extension. Both open the newest metadata.json in the bucket (Iceberg calls a table opened straight from its metadata file a static table):

File: open-lakehouse/read_iceberg.py

"""Read orders_raw as an Iceberg table with PyIceberg and DuckDB. No Spark involved."""

import duckdb
from common import S3_ENDPOINT, S3_KEY, S3_SECRET, s3
from pyiceberg.table import StaticTable

# The table's current Iceberg metadata is the newest metadata.json UniForm wrote.
listing = s3().list_objects_v2(Bucket="lakehouse", Prefix="tables/orders_raw/metadata/")
newest = max(
    (obj for obj in listing["Contents"] if obj["Key"].endswith(".metadata.json")),
    key=lambda obj: obj["LastModified"],
)
location = f"s3://lakehouse/{newest['Key']}"
print(f"Reading {location}\n")

table = StaticTable.from_metadata(location, properties={
    "s3.endpoint": S3_ENDPOINT,
    "s3.access-key-id": S3_KEY,
    "s3.secret-access-key": S3_SECRET,
    "s3.region": "us-east-1",
})
rows = table.scan(selected_fields=("event_id",)).to_arrow().num_rows
print(f"PyIceberg: {rows:,} rows (Iceberg v{table.format_version}, "
      f"converted from Delta version {table.properties['delta-version']})\n")

duck = duckdb.connect()
duck.sql("INSTALL iceberg; LOAD iceberg; INSTALL httpfs; LOAD httpfs;")
duck.sql(f"""
    CREATE SECRET (TYPE s3, KEY_ID '{S3_KEY}', SECRET '{S3_SECRET}', REGION 'us-east-1',
                   ENDPOINT '{S3_ENDPOINT.split("://")[1]}', URL_STYLE 'path', USE_SSL false)
""")
duck.sql(f"""
    SELECT event_date, count(*) AS events
    FROM iceberg_scan('{location}')
    GROUP BY event_date ORDER BY event_date
""").show()

python read_iceberg.py

Reading s3://lakehouse/tables/orders_raw/metadata/00000-01bfacd6-e573-475c-917f-34bc3a647658.metadata.json

PyIceberg: 34,021 rows (Iceberg v2, converted from Delta version 2)

┌────────────┬────────┐
│ event_date │ events │
│    date    │ int64  │
├────────────┼────────┤
│ 2026-09-01 │   4720 │
│ 2026-09-02 │   4382 │
│ 2026-09-03 │   5479 │
│ 2026-09-04 │   4508 │
│ 2026-09-05 │   5194 │
│ 2026-09-06 │   4962 │
│ 2026-09-07 │   4776 │
└────────────┴────────┘

The same rows per day as the Delta query in the catalog section, through Iceberg metadata. "Converted from Delta version 2" is the ALTER commit, and every later commit to the table gets new Iceberg metadata, so the readers keep up with the writes. (DuckDB downloads its iceberg and httpfs extensions the first time.)

Unity Catalog speaks Iceberg too: its Iceberg REST catalog, at /api/2.1/unity-catalog/iceberg with the catalog name as the warehouse, lists the catalog-managed UniForm tables. To hand one of them to an Iceberg client, the server reads the table's metadata file with its own S3 client, which UC 0.6.0 sets up for AWS S3, so on this laptop stack the readers open the metadata from the bucket directly instead.

Spark writes Delta here, and Iceberg only through UniForm. Native Iceberg writes from Spark 4.2 arrive with Iceberg's next release, which adds a Spark 4.2 runtime.

flowchart LR
  classDef new fill:#d97706,fill-opacity:0.72,stroke:#d97706,color:#ffffff
  classDef old fill:#4b5563,fill-opacity:0.72,stroke:#4b5563,color:#ffffff
  ib_SC["Spark Connect"]:::old
  subgraph ib_T["orders_raw"]
    ib_LOG["_delta_log"]:::old
    ib_PQ["Parquet<br/>data files"]:::old
    ib_MD["metadata/<br/>Iceberg"]:::new
  end
  ib_RD["PyIceberg<br/>DuckDB"]:::new
  ib_SC -->|"Delta commit"| ib_LOG
  ib_LOG -.->|"UniForm"| ib_MD
  ib_MD -.->|"lists"| ib_PQ
  ib_RD -->|"reads"| ib_MD

Pipelines: bronze, silver, gold with Spark Declarative Pipelines

So far every table was written by hand. In a real lakehouse the interesting tables are derived: cleaned events from raw ones, aggregates from cleaned ones. Spark Declarative Pipelines (SDP) lets you declare each table as a query; SDP works out the dependency graph from the tables each query reads and materializes them in order. Here it's the classic medallion layout:

Layer	Tables	What it does
bronze	`orders_bronze`, `locations`	the landing files, as they arrived
silver	`orders_silver`	each event once, with a known location, the JSON parsed and the city attached
gold	`revenue_by_brand`, `revenue_by_city_day`	the numbers people ask for

A pipeline is a spec file plus a folder of transformations. Make the folders:

mkdir -p pipeline/transformations

The spec names the catalog and schema the tables go to, and where SDP keeps its own state:

File: open-lakehouse/pipeline/spark-pipeline.yml

name: medallion
catalog: lakehouse
database: medallion
storage: s3://lakehouse/pipelines/medallion
libraries:
  - glob:
      include: transformations/**

Bronze reads files, so it's Python, the one place a query needs a DataFrame reader:

File: open-lakehouse/pipeline/transformations/bronze.py

"""Bronze: the landing files, as they arrived."""

from pyspark import pipelines as dp
from pyspark.sql import SparkSession

spark = SparkSession.active()

# Unity Catalog manages these tables: it picks their location under the
# catalog's storage root. These properties mark a Delta table as catalog-managed.
MANAGED = {
    "delta.feature.catalogManaged": "supported",
    "delta.checkpoint.writeStatsAsJson": "true",
    "delta.checkpoint.writeStatsAsStruct": "true",
}


@dp.materialized_view(format="delta", table_properties=MANAGED)
def orders_bronze():
    return spark.read.parquet("s3://lakehouse/landing/orders/")


@dp.materialized_view(format="delta", table_properties=MANAGED)
def locations():
    return spark.read.parquet("s3://lakehouse/landing/locations/")

Silver and gold are SQL. Each CREATE MATERIALIZED VIEW names the tables it reads, and that's all SDP needs to order them:

File: open-lakehouse/pipeline/transformations/silver.sql

-- Silver: each event once, with a known location, the order details parsed
-- out of the JSON body, and the city attached.
CREATE MATERIALIZED VIEW orders_silver
USING delta
TBLPROPERTIES ('delta.feature.catalogManaged' = 'supported',
               'delta.checkpoint.writeStatsAsJson' = 'true',
               'delta.checkpoint.writeStatsAsStruct' = 'true')
AS SELECT DISTINCT
  o.event_id,
  o.event_type,
  to_timestamp(o.ts) AS event_time,
  o.order_id,
  l.city,
  get_json_object(o.body, '$.brand') AS brand,
  CAST(get_json_object(o.body, '$.total') AS DOUBLE) AS total
FROM orders_bronze o
JOIN locations l ON o.location_id = l.location_id;

File: open-lakehouse/pipeline/transformations/gold.sql

-- Gold: the numbers people ask for, built from silver.
CREATE MATERIALIZED VIEW revenue_by_brand
USING delta
TBLPROPERTIES ('delta.feature.catalogManaged' = 'supported',
               'delta.checkpoint.writeStatsAsJson' = 'true',
               'delta.checkpoint.writeStatsAsStruct' = 'true')
AS SELECT
  brand,
  count(*) AS orders,
  round(sum(total), 2) AS revenue,
  round(avg(total), 2) AS avg_order
FROM orders_silver
WHERE event_type = 'order_created'
GROUP BY brand;

CREATE MATERIALIZED VIEW revenue_by_city_day
USING delta
TBLPROPERTIES ('delta.feature.catalogManaged' = 'supported',
               'delta.checkpoint.writeStatsAsJson' = 'true',
               'delta.checkpoint.writeStatsAsStruct' = 'true')
AS SELECT
  city,
  to_date(event_time) AS day,
  count(*) AS orders,
  round(sum(total), 2) AS revenue
FROM orders_silver
WHERE event_type = 'order_created'
GROUP BY city, to_date(event_time);

These are catalog-managed tables, so each one carries the three table properties that mark a Delta table as catalog-managed (Unity Catalog requires both checkpoint-stats properties when it creates one), and none gives a location. SDP also won't replace a table that already exists in Unity Catalog, so a run starts by clearing the previous run's tables:

File: open-lakehouse/prepare_medallion.py

"""Get lakehouse.medallion ready for a pipeline run: create it, or clear the last run.

SDP can't replace a table that already exists in Unity Catalog, so a rerun of the
pipeline starts by dropping the tables the previous run created.
"""

from common import spark

session = spark()
session.sql("CREATE SCHEMA IF NOT EXISTS lakehouse.medallion")
for row in session.sql("SHOW TABLES IN lakehouse.medallion").collect():
    session.sql(f"DROP TABLE lakehouse.medallion.{row.tableName}")
    print(f"dropped lakehouse.medallion.{row.tableName}")
print("lakehouse.medallion is ready")

Now prepare the schema, check the graph without writing anything, and run it. The pipeline CLI ships with pyspark-client; it runs on your machine, reads the spec and the transformations, and sends the graph to the Connect server, which does the work:

python prepare_medallion.py
SPARK_REMOTE=sc://localhost:15002 python -m pyspark.pipelines.cli dry-run --spec pipeline/spark-pipeline.yml
SPARK_REMOTE=sc://localhost:15002 python -m pyspark.pipelines.cli run --spec pipeline/spark-pipeline.yml

2026-10-01 14:36:58: Loading definitions. Root directory: '.../open-lakehouse/pipeline'.
2026-10-01 14:36:58: Found 3 files matching glob 'transformations/**/*'
...
2026-10-01 21:37:09: Flow lakehouse.medallion.orders_bronze has COMPLETED.
2026-10-01 21:37:09: Flow lakehouse.medallion.locations has COMPLETED.
2026-10-01 21:37:15: Flow lakehouse.medallion.orders_silver has COMPLETED.
2026-10-01 21:37:19: Flow lakehouse.medallion.revenue_by_city_day has COMPLETED.
2026-10-01 21:37:19: Flow lakehouse.medallion.revenue_by_brand has COMPLETED.
2026-10-01 21:37:21: Run is COMPLETED.

The dry run resolves every table and dependency before any data moves, which makes it the cheap thing to run first. Bronze's two tables run side by side, silver waits for both, and the two gold tables wait for silver. To check, read a gold table by name and list what the pipeline put in the catalog:

File: open-lakehouse/read_gold.py

"""Read a gold table by name, and list what the pipeline put in the catalog."""

from common import spark, uc

spark().sql("""
    SELECT * FROM lakehouse.medallion.revenue_by_brand ORDER BY revenue DESC
""").show()

for table in uc("GET", "/tables?catalog_name=lakehouse&schema_name=medallion")["tables"]:
    print(table["name"], table["table_type"], table["data_source_format"])

python read_gold.py

+-------------+------+---------+---------+
|        brand|orders|  revenue|avg_order|
+-------------+------+---------+---------+
|Sushi Express|  2140|160332.61|    74.92|
|  Curry House|  2219|133432.76|    60.13|
| Wok This Way|  2152|118668.13|    55.14|
| Pizza Planet|  2233| 93402.24|    41.83|
|    Taco Loco|  2168| 71548.53|     33.0|
+-------------+------+---------+---------+

locations MANAGED DELTA
orders_bronze MANAGED DELTA
orders_silver MANAGED DELTA
revenue_by_brand MANAGED DELTA
revenue_by_city_day MANAGED DELTA

The order counts are the distinct orders with a known location; the duplicates and the orders that lost their location stopped at silver. A managed table's location is the catalog's business, so the listing shows them as MANAGED. To run the pipeline again, run python prepare_medallion.py first.

flowchart LR
  classDef new fill:#d97706,fill-opacity:0.72,stroke:#d97706,color:#ffffff
  classDef old fill:#4b5563,fill-opacity:0.72,stroke:#4b5563,color:#ffffff
  pl_SPEC["spark-pipeline.yml<br/>bronze.py, *.sql"]:::new
  pl_SC["Spark Connect"]:::old
  subgraph pl_MD["lakehouse.medallion"]
    pl_BR[("bronze")]:::new
    pl_SI[("silver")]:::new
    pl_GO[("gold")]:::new
  end
  pl_SPEC -->|"graph"| pl_SC
  pl_SC --> pl_BR
  pl_BR --> pl_SI
  pl_SI --> pl_GO

Streaming: Real-Time Mode and micro-batches

The week of files is history; a stream is what's happening now. Kafka is the event log the orders flow through, and Spark's Structured Streaming reads it. By default Structured Streaming works in micro-batches: it collects whatever arrived since the last batch, processes it, commits, and starts over, so each record waits for the next batch. Real-Time Mode keeps the query's tasks running for a long batch instead and passes each record through the moment it arrives; the trigger's duration only sets how often the query writes a checkpoint. In Spark 4.2, Real-Time Mode runs stateless queries (filters, projections and the like, with no aggregation or deduplication), reads from Kafka and writes to Kafka, in the update output mode. Spark 4.2 is also the release that gave PySpark its trigger(realTime=...) keyword, so it works from Python over Spark Connect like everything else here.

This section builds two queries on the order stream: a stateless one in Real-Time Mode that writes clean orders back to Kafka, and a stateful one (deduplication and running totals) as an ordinary micro-batch query that writes into a Delta table in Unity Catalog.

Kafka

Add Kafka, under services::

File: open-lakehouse/compose.yaml, add under services:

  kafka:
    image: apache/kafka:4.2.0
    ports:
      - "${KAFKA_PORT:-9092}:9094"
    environment:
      KAFKA_NODE_ID: 1
      KAFKA_PROCESS_ROLES: broker,controller
      KAFKA_CONTROLLER_QUORUM_VOTERS: 1@kafka:9093
      KAFKA_CONTROLLER_LISTENER_NAMES: CONTROLLER
      KAFKA_LISTENERS: INTERNAL://:9092,CONTROLLER://:9093,EXTERNAL://:9094
      KAFKA_ADVERTISED_LISTENERS: INTERNAL://kafka:9092,EXTERNAL://localhost:${KAFKA_PORT:-9092}
      KAFKA_LISTENER_SECURITY_PROTOCOL_MAP: INTERNAL:PLAINTEXT,CONTROLLER:PLAINTEXT,EXTERNAL:PLAINTEXT
      KAFKA_INTER_BROKER_LISTENER_NAME: INTERNAL
      KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR: 1
      KAFKA_TRANSACTION_STATE_LOG_REPLICATION_FACTOR: 1
      KAFKA_TRANSACTION_STATE_LOG_MIN_ISR: 1
    healthcheck:
      test: /opt/kafka/bin/kafka-topics.sh --bootstrap-server localhost:9092 --list
      interval: 10s
      retries: 30

This is the official Apache Kafka image in KRaft mode: one process is both the broker and its own controller, with no ZooKeeper. It has two listeners because its clients are in two places. Spark, in the Compose network, connects to kafka:9092; your machine connects to localhost:9092, which Docker forwards to the EXTERNAL listener on container port 9094, and that listener tells clients to come back to localhost.

docker compose up -d --wait

Create the input topic and the Real-Time Mode query's output topic. orders gets two partitions, so the Real-Time Mode query reads it with two parallel tasks:

docker compose exec kafka /opt/kafka/bin/kafka-topics.sh --bootstrap-server kafka:9092 --create --topic orders --partitions 2
docker compose exec kafka /opt/kafka/bin/kafka-topics.sh --bootstrap-server kafka:9092 --create --topic orders-clean --partitions 2

Created topic orders.
Created topic orders-clean.

A stateless query in Real-Time Mode

The first query keeps the events a downstream service would act on: new orders with a known location. It pulls the brand and total out of the JSON body and writes each one to orders-clean, along with the time the input reached Kafka, so you can time it later. It's an ordinary streaming query with a different trigger:

File: open-lakehouse/stream_clean.py

"""Real-Time Mode: clean each order event from Kafka into another topic as it arrives."""

from common import spark
from pyspark.sql import functions as f

EVENT = "event_id STRING, event_type STRING, ts STRING, location_id INT, order_id STRING, body STRING"

session = spark()
orders = (
    session.readStream.format("kafka")
    .option("kafka.bootstrap.servers", "kafka:9092")  # Kafka as the cluster sees it
    .option("subscribe", "orders")
    .load()
    .select(f.from_json(f.col("value").cast("string"), EVENT).alias("e"), "timestamp")
)

clean = orders.where("e.event_type = 'order_created' AND e.location_id IS NOT NULL").select(
    f.col("e.order_id").alias("key"),
    f.to_json(f.struct(
        "e.event_id",
        "e.location_id",
        f.get_json_object("e.body", "$.brand").alias("brand"),
        f.get_json_object("e.body", "$.total").cast("double").alias("total"),
        f.unix_millis("timestamp").alias("received_ms"),  # when it reached Kafka
    )).alias("value"),
)

query = (
    clean.writeStream.queryName("orders_clean")
    .format("kafka")
    .option("kafka.bootstrap.servers", "kafka:9092")
    .option("topic", "orders-clean")
    .option("checkpointLocation", "/opt/spark/work-dir/checkpoints/orders_clean")
    .outputMode("update")  # the output mode Real-Time Mode supports
    .trigger(realTime="5 minutes")  # how often to checkpoint, not a latency target
    .start()
)
print("orders -> orders-clean, in Real-Time Mode. Ctrl+C to stop.")
try:
    query.awaitTermination()
except KeyboardInterrupt:
    query.stop()

The query runs on the Connect server, so it reaches Kafka as kafka:9092, its name inside the Compose network. Its checkpoint goes in the spark-work volume that the Connect server and the worker share. (SeaweedFS's S3 API handles everything Delta writes, but Spark's streaming checkpoints don't start on it: Spark expects a new checkpoint's offsets folder to list as empty, and SeaweedFS lists the folder's own marker object.)

A stateful query into Delta

The second query keeps a running count of orders and the revenue for each kitchen location. Two stateful steps do the work: dropDuplicates remembers the event ids it has seen and drops repeats, because a duplicate order would otherwise count twice, and groupBy(...).agg(...) keeps the running totals. Real-Time Mode in Spark 4.2 doesn't run stateful queries, so this one is a micro-batch query, once a second:

File: open-lakehouse/stream_totals.py

"""Micro-batch: deduplicate orders and keep running totals per location, in Delta."""

from common import spark
from pyspark.sql import functions as f

EVENT = "event_id STRING, event_type STRING, ts STRING, location_id INT, order_id STRING, body STRING"

session = spark()
session.conf.set("spark.sql.shuffle.partitions", "4")
session.sql("""
    CREATE TABLE IF NOT EXISTS lakehouse.tutorial.live_totals (
        location_id INT, orders BIGINT, revenue DOUBLE, last_order_at TIMESTAMP
    ) USING delta LOCATION 's3://lakehouse/tables/live_totals'
""")

totals = (
    session.readStream.format("kafka")
    .option("kafka.bootstrap.servers", "kafka:9092")
    .option("subscribe", "orders")
    .load()
    .select(f.from_json(f.col("value").cast("string"), EVENT).alias("e"))
    .select("e.*", f.to_timestamp("e.ts").alias("event_time"))
    .where("event_type = 'order_created' AND location_id IS NOT NULL")
    .withWatermark("event_time", "10 minutes")  # how long to remember an event id
    .dropDuplicates(["event_id", "event_time"])
    .groupBy("location_id")
    .agg(
        f.count("*").alias("orders"),
        f.round(f.sum(f.get_json_object("body", "$.total").cast("double")), 2).alias("revenue"),
        f.max("event_time").alias("last_order_at"),
    )
)

query = (
    totals.writeStream.queryName("live_totals")
    .format("delta")
    .outputMode("complete")  # each batch rewrites the four rows with the current totals
    .option("checkpointLocation", "/opt/spark/work-dir/checkpoints/live_totals")
    .trigger(processingTime="1 second")
    .toTable("lakehouse.tutorial.live_totals")
)
print("orders -> lakehouse.tutorial.live_totals, every second. Ctrl+C to stop.")
try:
    query.awaitTermination()
except KeyboardInterrupt:
    query.stop()

A few details matter here. The filter comes before the deduplication, so the state only holds events that count. The watermark bounds that state: without it, dropDuplicates would keep every event id forever, and with it, ids more than 10 minutes (in event time) behind the newest event are dropped from state, which is also why event_time is part of the deduplication key. The generator resends its duplicates right away and sends a few events three minutes late, so 10 minutes covers both. The complete output mode suits a result this small: every batch rewrites four rows, so the table always holds the current totals, and each batch is a Delta commit you can time-travel to.

Run them

Start each query in a terminal of its own (source .venv/bin/activate first), and leave them running:

python stream_clean.py

python stream_totals.py

With both waiting, stream a day of orders into Kafka from a third terminal. It sends 50 events a second and stops by itself after about two minutes:

python generate_orders.py stream

sent 1,000 events
...
sent 4,625 events to the topic orders

Here are a few records as the Real-Time Mode query wrote them:

docker compose exec kafka /opt/kafka/bin/kafka-console-consumer.sh --bootstrap-server kafka:9092 --topic orders-clean --from-beginning --max-messages 3

{"event_id":"4de57cb207-0","location_id":4,"brand":"Curry House","total":23.98,"received_ms":1790890668647}
{"event_id":"904638cd9b-0","location_id":2,"brand":"Wok This Way","total":23.63,"received_ms":1790890668769}
{"event_id":"bdbede7a39-0","location_id":3,"brand":"Pizza Planet","total":33.93,"received_ms":1790890668890}
Processed a total of 3 messages

To check both queries, compare their output with the input. This script reads the topics once, as batches: it counts the order_created events with a location two ways (all records, and distinct event ids), reads the current totals from the Delta table, and measures how long each record spent inside the Real-Time Mode query, from its input's Kafka timestamp to its output's:

File: open-lakehouse/check_stream.py

"""Check both streaming queries against what the generator sent."""

from common import spark
from pyspark.sql import functions as f

EVENT = "event_id STRING, event_type STRING, ts STRING, location_id INT, order_id STRING, body STRING"

session = spark()


def topic(name):
    """Everything in a Kafka topic so far, read once as a batch."""
    return (
        session.read.format("kafka")
        .option("kafka.bootstrap.servers", "kafka:9092")
        .option("subscribe", name)
        .option("startingOffsets", "earliest")
        .load()
    )


sent = (
    topic("orders")
    .select(f.from_json(f.col("value").cast("string"), EVENT).alias("e"))
    .where("e.event_type = 'order_created' AND e.location_id IS NOT NULL")
    .agg(f.count("*").alias("records"), f.count_distinct("e.event_id").alias("distinct"))
    .first()
)
totals = session.table("lakehouse.tutorial.live_totals").orderBy("location_id")
totals.show(truncate=False)

print(f"order_created records sent:        {sent['records']:,}")
print(f"distinct order_created event ids:  {sent['distinct']:,}")
print(f"orders in live_totals:             {totals.agg(f.sum('orders')).first()[0]:,}")

# How long each record spent in the Real-Time Mode query, Kafka to Kafka.
spent = topic("orders-clean").select(
    (f.unix_millis("timestamp")
     - f.get_json_object(f.col("value").cast("string"), "$.received_ms").cast("long")).alias("ms")
)
p50, p99 = spent.agg(f.percentile_approx("ms", [0.5, 0.99])).first()[0]
print(f"\norders-clean: {spent.count():,} records, p50 {p50} ms, p99 {p99} ms in Spark")

python check_stream.py

+-----------+------+--------+-----------------------+
|location_id|orders|revenue |last_order_at          |
+-----------+------+--------+-----------------------+
|1          |361   |18666.46|2026-10-01 21:39:19.782|
|2          |354   |19276.01|2026-10-01 21:39:19.58 |
|3          |380   |19758.75|2026-10-01 21:39:19.802|
|4          |384   |19255.33|2026-10-01 21:39:19.883|
+-----------+------+--------+-----------------------+

order_created records sent:        1,512
distinct order_created event ids:  1,479
orders in live_totals:             1,479

orders-clean: 1,512 records, p50 1 ms, p99 3 ms in Spark

The totals match the distinct count, not the raw one: the duplicate orders the generator sent never made it into a total. (Check too early, while the stateful query is still catching up, and its total trails the distinct count for a second.) The Real-Time Mode query, which has no state, passed every record through, duplicates included, in a millisecond or two each. A micro-batch query with a one-second trigger holds each record until its batch runs, half a second on average, before it does anything with it. In Spark 4.3, Real-Time Mode also runs stateful queries.

Stop both queries with Ctrl+C in their terminals; each script catches it and stops its query on the server.

flowchart LR
  classDef new fill:#d97706,fill-opacity:0.72,stroke:#d97706,color:#ffffff
  classDef old fill:#4b5563,fill-opacity:0.72,stroke:#4b5563,color:#ffffff
  ks_GEN["generator"]:::new
  ks_K["Kafka :9092<br/>topic orders"]:::new
  subgraph ks_SP["Spark 4.2"]
    ks_SL["stateless<br/>Real-Time Mode"]:::new
    ks_SF["stateful<br/>micro-batch"]:::new
  end
  ks_OUT["Kafka<br/>orders-clean"]:::new
  ks_T[("lakehouse.tutorial<br/>live_totals")]:::new
  ks_GEN --> ks_K
  ks_K --> ks_SL
  ks_K --> ks_SF
  ks_SL -->|"milliseconds"| ks_OUT
  ks_SF -->|"every second"| ks_T

Orchestration: Airflow runs the pipeline and the maintenance

Everything so far ran because you typed it. Apache Airflow runs it on a schedule, retries it, and keeps a record of every run. It doesn't touch the data itself: its tasks talk to the same Connect server you've been using, so Airflow needs only pyspark-client, not a JVM. Two DAGs, in a dags folder:

mkdir dags

The first runs the medallion pipeline every day. It does what you did by hand: clears the previous run's tables, runs the spec through SparkPipelinesOperator from Airflow's Spark provider, and then checks that the gold table isn't empty. With a spark_connect connection, the operator runs the same pipeline CLI you ran, inside the task, against the Connect server:

File: open-lakehouse/dags/medallion.py

"""Run the medallion pipeline every day, on the cluster, through Spark Connect."""

from datetime import datetime

from airflow.providers.apache.spark.operators.spark_pipelines import SparkPipelinesOperator
from airflow.sdk import DAG, task

with DAG("medallion", schedule="@daily", start_date=datetime(2026, 9, 1), catchup=False):

    @task
    def prepare():
        """Create the schema, or drop the last run's tables (SDP won't replace them)."""
        from pyspark.sql import SparkSession

        spark = SparkSession.builder.remote("sc://spark-connect:15002").getOrCreate()
        spark.sql("CREATE SCHEMA IF NOT EXISTS lakehouse.medallion")
        for row in spark.sql("SHOW TABLES IN lakehouse.medallion").collect():
            spark.sql(f"DROP TABLE lakehouse.medallion.{row.tableName}")

    run_pipeline = SparkPipelinesOperator(
        task_id="run_pipeline",
        pipeline_spec="/opt/airflow/pipeline/spark-pipeline.yml",
        pipeline_command="run",
        conn_id="spark_connect_default",
    )

    @task
    def check_gold():
        from pyspark.sql import SparkSession

        spark = SparkSession.builder.remote("sc://spark-connect:15002").getOrCreate()
        brands = spark.table("lakehouse.medallion.revenue_by_brand").count()
        print(f"revenue_by_brand has {brands} rows")
        if brands == 0:
            raise ValueError("the gold table is empty")

    prepare() >> run_pipeline >> check_gold()

The second compacts the two external tables every night and deletes the files no version from the last week needs. Compaction matters for both: the batch table was written as a few small files, and every micro-batch rewrites the streaming table as a few more.

File: open-lakehouse/dags/maintenance.py

"""Every night, compact the external Delta tables and clean up files nothing needs."""

from datetime import datetime

from airflow.sdk import DAG, task

TABLES = ["lakehouse.tutorial.orders_raw", "lakehouse.tutorial.live_totals"]

with DAG("delta_maintenance", schedule="0 3 * * *", start_date=datetime(2026, 9, 1), catchup=False):

    @task
    def optimize_and_vacuum():
        from pyspark.sql import SparkSession

        spark = SparkSession.builder.remote("sc://spark-connect:15002").getOrCreate()
        for table in TABLES:
            before = spark.sql(f"DESCRIBE DETAIL {table}").first()["numFiles"]
            spark.sql(f"OPTIMIZE {table}").collect()  # many small files into a few big ones
            spark.sql(f"VACUUM {table} RETAIN 168 HOURS").collect()  # keep a week of history
            after = spark.sql(f"DESCRIBE DETAIL {table}").first()["numFiles"]
            print(f"{table}: {before} files -> {after}")

    optimize_and_vacuum()

It leaves the pipeline's tables alone: for catalog-managed tables the catalog owns maintenance, and Delta doesn't run client-side OPTIMIZE and VACUUM on them.

Add Airflow, under services::

File: open-lakehouse/compose.yaml, add under services:

  airflow:
    build:
      dockerfile_inline: |
        FROM apache/airflow:3.3.2-python3.12
        RUN pip install --no-cache-dir apache-airflow==3.3.2 apache-airflow-providers-apache-spark==6.3.2 pyspark-client==4.2.0
    command: standalone
    environment:
      AIRFLOW__CORE__LOAD_EXAMPLES: "false"
      AIRFLOW__CORE__SIMPLE_AUTH_MANAGER_ALL_ADMINS: "true"
      AIRFLOW_CONN_SPARK_CONNECT_DEFAULT: '{"conn_type": "spark_connect", "host": "spark-connect", "port": 15002}'
    ports:
      - "${AIRFLOW_PORT:-8085}:8080"
    volumes:
      - airflow-data:/opt/airflow
      - ./dags:/opt/airflow/dags
      - ./pipeline:/opt/airflow/pipeline
    healthcheck:
      test: curl -sf http://localhost:8080/api/v2/monitor/health
      interval: 5s
      start_period: 5m

and its volume, under volumes::

File: open-lakehouse/compose.yaml, add under volumes:

  airflow-data:

The build section is a two-line Dockerfile inside compose.yaml: the official Airflow image plus the Spark provider and pyspark-client, built the first time you start it. airflow standalone runs every Airflow component in one container with a SQLite database, which is the simplest setup for one laptop. The airflow-data volume keeps that database and the task logs when the container is recreated. The connection comes from an environment variable, and SIMPLE_AUTH_MANAGER_ALL_ADMINS skips the login page, which is fine for a stack that only listens on your machine. The DAGs and the pipeline folder are mounted in, so the operator finds the same spec you ran.

docker compose up -d --wait

DAGs start paused. Unpause both; the scheduler then starts each one's most recent scheduled run right away:

docker compose exec airflow airflow dags unpause medallion
docker compose exec airflow airflow dags unpause delta_maintenance

Give them a minute or two, then list the runs:

docker compose exec airflow airflow dags list-runs medallion -o plain
docker compose exec airflow airflow dags list-runs delta_maintenance -o plain

dag_id     run_id                                state    run_after                  logical_date               start_date                        end_date
medallion  scheduled__2026-10-01T00:00:00+00:00  success  2026-10-01T00:00:00+00:00  2026-10-01T00:00:00+00:00  2026-10-01T21:40:07.659874+00:00  2026-10-01T21:40:49.719529+00:00
dag_id             run_id                                state    run_after                  logical_date               start_date                        end_date
delta_maintenance  scheduled__2026-10-01T03:00:00+00:00  success  2026-10-01T03:00:00+00:00  2026-10-01T03:00:00+00:00  2026-10-01T21:40:09.561521+00:00  2026-10-01T21:40:41.881096+00:00

and read the results out of the task logs:

docker compose exec airflow grep -rhoE "revenue_by_brand has [0-9]+ rows|lakehouse[.a-z_]+: [0-9]+ files -> [0-9]+" /opt/airflow/logs

lakehouse.tutorial.orders_raw: 3 files -> 1
lakehouse.tutorial.live_totals: 2 files -> 1
revenue_by_brand has 5 rows

The pipeline rebuilt the medallion tables, and the maintenance DAG compacted each external table into one file. orders_raw has UniForm on, so the compaction was converted to Iceberg too; read it as Iceberg again, and the readers find the new snapshot and the same rows:

python read_iceberg.py

Reading s3://lakehouse/tables/orders_raw/metadata/00000-47aecc6b-89d1-40f8-97eb-b8c390180aca.metadata.json

PyIceberg: 34,021 rows (Iceberg v2, converted from Delta version 3)
...

The Airflow UI at http://localhost:8085 shows the same runs in its grid view, with every task's log.

flowchart LR
  classDef new fill:#d97706,fill-opacity:0.72,stroke:#d97706,color:#ffffff
  classDef old fill:#4b5563,fill-opacity:0.72,stroke:#4b5563,color:#ffffff
  af_AF["Airflow :8085<br/>two DAGs"]:::new
  af_SC["Spark Connect"]:::old
  af_UC["Unity Catalog"]:::old
  af_T[("medallion and<br/>tutorial tables")]:::old
  af_AF -->|"SDP run, OPTIMIZE"| af_SC
  af_SC -->|"names"| af_UC
  af_SC --> af_T

ML: MLflow tracks what you compute

The last layer is where the tables get used. MLflow records runs of whatever you compute from them, whether that's model training or a scheduled metric: parameters, metrics and artifacts, all searchable later. Here the tracking server keeps its metadata in SQLite and its artifacts in the bucket, and clients upload artifacts through the server, so only the server needs the S3 credentials.

Add MLflow, under services::

File: open-lakehouse/compose.yaml, add under services:

  mlflow:
    build:
      dockerfile_inline: |
        FROM ghcr.io/mlflow/mlflow:v3.16.1
        RUN pip install --no-cache-dir boto3==1.43.93
    command: >
      mlflow server --host 0.0.0.0 --port 5000 --workers 1
      --backend-store-uri sqlite:////mlflow/mlflow.db
      --artifacts-destination s3://lakehouse/mlflow
    environment:
      MLFLOW_S3_ENDPOINT_URL: http://seaweedfs:8333
      AWS_ACCESS_KEY_ID: lakehouse
      AWS_SECRET_ACCESS_KEY: lakehouse-secret
    ports:
      - "${MLFLOW_PORT:-5000}:5000"
    volumes:
      - mlflow-data:/mlflow
    healthcheck:
      test: python -c "import urllib.request; urllib.request.urlopen('http://localhost:5000/health')"
      interval: 5s
      retries: 30

and its volume, under volumes::

File: open-lakehouse/compose.yaml, add under volumes:

  mlflow-data:

The official image doesn't include boto3, which the server needs to write to S3, so the build adds it. --artifacts-destination is where the server puts artifacts it receives from clients.

docker compose up -d --wait

This is a PySpark job that logs a run: it aggregates the gold table on the cluster, logs the table it read, three metrics and the table as a CSV artifact, then reads the run back from the tracking server and lists the artifact in the bucket:

File: open-lakehouse/log_run.py

"""Log an MLflow run from a PySpark job, then read it back from the tracking server."""

import mlflow
from common import MLFLOW_TRACKING_URI, s3, spark
from pyspark.sql import functions as f

gold = spark().table("lakehouse.medallion.revenue_by_brand")

mlflow.set_tracking_uri(MLFLOW_TRACKING_URI)
mlflow.set_experiment("ghost-kitchen")
with mlflow.start_run(run_name="brand-revenue") as run:
    stats = gold.agg(
        f.count("*").alias("brands"),
        f.round(f.sum("revenue"), 2).alias("revenue"),
        f.max("avg_order").alias("best_avg_order"),
    ).first()
    mlflow.log_param("source_table", "lakehouse.medallion.revenue_by_brand")
    mlflow.log_metrics(stats.asDict())
    mlflow.log_text(gold.toPandas().to_csv(index=False), "revenue_by_brand.csv")

logged = mlflow.get_run(run.info.run_id)
print(f"{logged.info.run_name}: {logged.info.status}")
for name, value in sorted(logged.data.metrics.items()):
    print(f"  {name} = {value}")

# The artifact went through the tracking server into the bucket.
for obj in s3().list_objects_v2(Bucket="lakehouse", Prefix="mlflow/")["Contents"]:
    print(obj["Key"])

python log_run.py

2026/10/01 14:41:30 INFO mlflow.tracking.fluent: Experiment with name 'ghost-kitchen' does not exist. Creating a new experiment.
🏃 View run brand-revenue at: http://localhost:5000/#/experiments/1/runs/88457b52f19d43fea413f804bf7d6509
🧪 View experiment at: http://localhost:5000/#/experiments/1
brand-revenue: FINISHED
  best_avg_order = 74.92
  brands = 5.0
  revenue = 577384.27
mlflow/1/88457b52f19d43fea413f804bf7d6509/artifacts/revenue_by_brand.csv

The run, its metrics and the artifact are also in the MLflow UI at http://localhost:5000.

flowchart LR
  classDef new fill:#d97706,fill-opacity:0.72,stroke:#d97706,color:#ffffff
  classDef old fill:#4b5563,fill-opacity:0.72,stroke:#4b5563,color:#ffffff
  ml_JOB["PySpark job"]:::new
  ml_SC["Spark Connect"]:::old
  ml_ML["MLflow :5000"]:::new
  ml_SW[("SeaweedFS")]:::old
  ml_JOB -->|"reads gold"| ml_SC
  ml_JOB -->|"logs run"| ml_ML
  ml_ML -->|"artifacts"| ml_SW

You built it!

The open lakehouse on Spark 4.2, as built in this post: your Python scripts, Airflow standalone and MLflow on top, all reaching one Spark Connect server; the generator feeding the landing files and the Kafka orders topic in KRaft mode; inside Spark 4.2, a Real-Time Mode query writing orders-clean back to Kafka and a stateful micro-batch query writing totals into Delta, beside the SDP medallion and batch writes; Unity Catalog OSS and SeaweedFS below, with PyIceberg and DuckDB reading the UniForm table's Iceberg metadata from SeaweedFS

Reading it from the top: your scripts, Airflow's DAGs and your MLflow runs all reach Spark through one Connect endpoint. The generator feeds both the landing files in the bucket and the orders topic. Spark turns that topic into clean events within milliseconds in Real-Time Mode, and into deduplicated running totals in a Delta table every second. It also runs the declarative pipeline, writes batches and compacts tables; Unity Catalog names every table and knows where it lives; SeaweedFS holds every file; and with UniForm on, Iceberg engines read the same files as Delta.

Here's the whole compose.yaml you built:

File: open-lakehouse/compose.yaml

services:
  seaweedfs:
    image: chrislusf/seaweedfs:3.80
    entrypoint: weed
    command: server -dir=/data -s3 -s3.config=/etc/seaweedfs/s3.json -volume.max=0 -master.volumeSizeLimitMB=1024
    ports:
      - "${S3_PORT:-8333}:8333"
    volumes:
      - ./s3.json:/etc/seaweedfs/s3.json:ro
      - seaweedfs-data:/data
    healthcheck:
      test: wget -qO /dev/null http://127.0.0.1:8333/status
      interval: 5s
      retries: 30

  spark-master:
    image: apache/spark:4.2.0
    hostname: spark-master
    command: /opt/spark/bin/spark-class org.apache.spark.deploy.master.Master --host spark-master
    ports:
      - "${SPARK_UI_PORT:-8080}:8080"
    healthcheck:
      test: curl -sf http://localhost:8080 > /dev/null
      interval: 5s
      retries: 30

  spark-worker:
    image: apache/spark:4.2.0
    hostname: spark-worker
    command: /opt/spark/bin/spark-class org.apache.spark.deploy.worker.Worker --host spark-worker spark://spark-master:7077
    environment:
      SPARK_WORKER_CORES: "6"
      SPARK_WORKER_MEMORY: 4g
    volumes:
      - spark-work:/opt/spark/work-dir
    depends_on:
      spark-master:
        condition: service_healthy

  spark-connect:
    image: apache/spark:4.2.0
    hostname: spark-connect
    command: >
      /opt/spark/sbin/start-connect-server.sh
      --master spark://spark-master:7077
      --packages io.delta:delta-spark_4.2_2.13:4.4.0,io.delta:delta-iceberg_2.13:4.4.0,io.unitycatalog:unitycatalog-spark_4.2_2.13:0.6.0,org.apache.hadoop:hadoop-aws:3.5.0,org.apache.spark:spark-sql-kafka-0-10_2.13:4.2.0
      --exclude-packages io.delta:delta-spark_4.1_2.13
    environment:
      SPARK_NO_DAEMONIZE: "1"
    ports:
      - "${SPARK_CONNECT_PORT:-15002}:15002"
    volumes:
      - ./spark-conf:/opt/spark/conf
      - spark-work:/opt/spark/work-dir
    depends_on:
      spark-master:
        condition: service_healthy
    healthcheck:
      test: ["CMD", "bash", "-c", "echo > /dev/tcp/localhost/15002"]
      interval: 5s
      start_period: 15m

  unity-catalog:
    image: unitycatalog/unitycatalog:v0.6.0
    ports:
      - "${UC_PORT:-8081}:8080"
    volumes:
      - ./server.properties:/home/unitycatalog/etc/conf/server.properties:ro
      - uc-data:/home/unitycatalog/etc/db
    healthcheck:
      test: wget -qO /dev/null http://127.0.0.1:8080/api/2.1/unity-catalog/catalogs
      interval: 5s
      retries: 30

  kafka:
    image: apache/kafka:4.2.0
    ports:
      - "${KAFKA_PORT:-9092}:9094"
    environment:
      KAFKA_NODE_ID: 1
      KAFKA_PROCESS_ROLES: broker,controller
      KAFKA_CONTROLLER_QUORUM_VOTERS: 1@kafka:9093
      KAFKA_CONTROLLER_LISTENER_NAMES: CONTROLLER
      KAFKA_LISTENERS: INTERNAL://:9092,CONTROLLER://:9093,EXTERNAL://:9094
      KAFKA_ADVERTISED_LISTENERS: INTERNAL://kafka:9092,EXTERNAL://localhost:${KAFKA_PORT:-9092}
      KAFKA_LISTENER_SECURITY_PROTOCOL_MAP: INTERNAL:PLAINTEXT,CONTROLLER:PLAINTEXT,EXTERNAL:PLAINTEXT
      KAFKA_INTER_BROKER_LISTENER_NAME: INTERNAL
      KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR: 1
      KAFKA_TRANSACTION_STATE_LOG_REPLICATION_FACTOR: 1
      KAFKA_TRANSACTION_STATE_LOG_MIN_ISR: 1
    healthcheck:
      test: /opt/kafka/bin/kafka-topics.sh --bootstrap-server localhost:9092 --list
      interval: 10s
      retries: 30

  airflow:
    build:
      dockerfile_inline: |
        FROM apache/airflow:3.3.2-python3.12
        RUN pip install --no-cache-dir apache-airflow==3.3.2 apache-airflow-providers-apache-spark==6.3.2 pyspark-client==4.2.0
    command: standalone
    environment:
      AIRFLOW__CORE__LOAD_EXAMPLES: "false"
      AIRFLOW__CORE__SIMPLE_AUTH_MANAGER_ALL_ADMINS: "true"
      AIRFLOW_CONN_SPARK_CONNECT_DEFAULT: '{"conn_type": "spark_connect", "host": "spark-connect", "port": 15002}'
    ports:
      - "${AIRFLOW_PORT:-8085}:8080"
    volumes:
      - airflow-data:/opt/airflow
      - ./dags:/opt/airflow/dags
      - ./pipeline:/opt/airflow/pipeline
    healthcheck:
      test: curl -sf http://localhost:8080/api/v2/monitor/health
      interval: 5s
      start_period: 5m

  mlflow:
    build:
      dockerfile_inline: |
        FROM ghcr.io/mlflow/mlflow:v3.16.1
        RUN pip install --no-cache-dir boto3==1.43.93
    command: >
      mlflow server --host 0.0.0.0 --port 5000 --workers 1
      --backend-store-uri sqlite:////mlflow/mlflow.db
      --artifacts-destination s3://lakehouse/mlflow
    environment:
      MLFLOW_S3_ENDPOINT_URL: http://seaweedfs:8333
      AWS_ACCESS_KEY_ID: lakehouse
      AWS_SECRET_ACCESS_KEY: lakehouse-secret
    ports:
      - "${MLFLOW_PORT:-5000}:5000"
    volumes:
      - mlflow-data:/mlflow
    healthcheck:
      test: python -c "import urllib.request; urllib.request.urlopen('http://localhost:5000/health')"
      interval: 5s
      retries: 30

volumes:
  seaweedfs-data:
  spark-work:
  uc-data:
  airflow-data:
  mlflow-data:

docker compose ps shows the whole stack at once:

docker compose ps --format "table {{.Service}}\t{{.Status}}"

SERVICE         STATUS
airflow         Up 37 seconds (healthy)
kafka           Up 4 minutes (healthy)
mlflow          Up 37 seconds (healthy)
seaweedfs       Up 6 minutes (healthy)
spark-connect   Up 5 minutes (healthy)
spark-master    Up 6 minutes (healthy)
spark-worker    Up 6 minutes
unity-catalog   Up 5 minutes (healthy)

To stop for the day, docker compose stop, and docker compose up -d --wait brings it all back. docker compose down removes the containers but keeps the named volumes, so your tables, the catalog and the downloaded JARs survive; docker compose down -v removes the volumes too and starts you over.

Where to go next

This post builds the smallest version of the stack that teaches each layer. open-lakehouse is the complete, maintained version of it: the same layers driven by one CLI, with CI, more demos (a Delta deep dive, Unity Catalog governance, a longer streaming benchmark, MLflow's model registry), Delta Sharing and a web dashboard. And if you want the story of the workbench this stack grew out of, that's in Hosting an Entire Open Lakehouse on Port 8080.

DEV Community: lnc

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

Usage

Where this code lives

The program

The same call, from Spark

Exploring pluggability

Where formats come into play and the compatibility kit

Adding a format: the Vortex example

What a format has to pass

Inside the kit

Implications for native multimodal in Iceberg

FILE

VECTOR

What the two proposals have in common

State of the work and future

Keeping up with the Joneses: What your next step is

Further reading

An Idiot's Guide to Spark Connect Clients

How Spark works

What the shared driver costs you

Connecting from Python

Before and after: setting up a session

Before and after: the rest of your code

Running a Connect server

Running a job

What you need locally

Clients in other languages

Debugging

The whole guide, on one page

Spark's Modules, Seen Through Spark Connect

What "reachable over Connect" actually means

Built for it: SQL and Declarative Pipelines

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

Structured Streaming and Real-Time Mode

MLlib on Connect

Python UDFs on Arrow

What crossing the boundary cost me

Left behind: GraphX, SparkR, and the RDD APIs

Scheduling SDP: `SparkPipelinesOperator`