Can we bring SIMD-style execution directly into Spark’s JVM execution path, using Java itself?

That is the idea behind VecRuntime, a JVM-native vectorized execution engine for Spark SQL. The project uses the Java Vector API to execute Spark SQL operators over Arrow-layout columnar batches. Unsupported operations simply fall back to Spark like on Comet.

Consider a completely ordinary PySpark query:

df = (
    sales
    .where(...)
    .groupBy(...)
    .agg(...)
    .orderBy(...)
)

There is nothing VecRuntime specific here. The application still uses the standard DataFrame API. The difference happens later.

Spark generates a physical plan and VecRuntime applies a planning rule that replaces supported Spark operators with vectorized implementations.

Anything that cannot be converted remains a normal Spark operator. That fallback model is important because it means acceleration can be incremental rather than all-or-nothing.

Why the JVM?

When people hear “vectorized execution”, they often think of native code. That is a reasonable assumption. Modern analytical engines frequently use C++, SIMD intrinsics, native memory layouts, and specialized runtimes. VecRuntime takes a different route. Its kernels are implemented in Java using the Java Vector API. The Java Vector API allows code to operate on multiple values per instruction while still remaining inside the JVM.

On an ARM64 machine, the JVM can use the available NEON or SVE vector instructions. On machines with wider x86 SIMD capabilities, the Vector API can select wider species automatically like AVX512.

What does VecRuntime accelerate?

The project currently implements vectorized versions of several core Spark SQL operators covering TPC-H and TPC-DS queries (details here).

However, the important part is not simply the number of operators. It is the composability. A vectorized Filter can feed a vectorized Aggregate, which can feed another vectorized operator, without converting back to Spark rows between every step.

That is where the architecture starts becoming interesting.

The columnar representation

VecRuntime operates on columnar batches using an Arrow-compatible layout. That means an execution pipeline can look like:

The operators work directly on columns rather than repeatedly materializing UnsafeRow objects. This matters because a vectorized operator is only useful if the surrounding pipeline preserves the vectorized representation. And that leads to one of the biggest lessons from the project.

The hidden cost: shuffle boundaries

Suppose we have:

VectorAggregate -> Spark Shuffle -> VectorAggregate

Spark’s traditional shuffle machinery is fundamentally row-oriented. That means a columnar batch can end up going through a sequence conceptually similar to:

The vectorized operator may be faster, but the conversion overhead can erase much of the benefit. This is why the example repository contains an optional columnar shuffle.

A shuffle designed for columnar execution

The VecRuntime shuffle moves Arrow record batches rather than rows. The architecture is roughly:

For remote shuffle, the current implementation uses Apache Arrow Flight over gRPC. The important property is not the transport itself.

The data can remain columnar across the shuffle boundary.

That removes the row-to-columnar conversion that otherwise happens at every stage boundary.

The PySpark example

The example repository was built specifically to demonstrate this approach without requiring a large Spark deployment.

The repository uses uv to manage the Python environment ( you will need Java 25too). The quick start is simply:

The repository uses uv to manage the Python environment.

The quick start is simply:

git clone https://github.com/vecruntime/pyspark-example
cd pyspark-example
uv run example.py

The first execution generates a small star schema:

sales
  ~20 million rows
stores
  1,000 rows

The generated data is written as Parquet and reused across subsequent runs. This gives us a simple but realistic analytical workload without needing an external dataset.

The example contains three DataFrame queries.

1. A TPC-H Q1-style aggregation

The first query filters rows, computes derived expressions, performs a grouped aggregation and orders the result. Conceptually:

Parquet scan
     ↓
Filter
     ↓
HashAggregate
     ↓
Sort

This is a particularly good shape for vectorization because it contains arithmetic, predicates and aggregation over large numbers of rows.

2. A broadcast join and top-N query

The second query joins the large sales table with the small stores dimension:

sales
   │
 Filter
   │
   ├──────── Broadcast stores
   │
 BroadcastHashJoin
   │
 Aggregate
   │
 TakeOrdered / LIMIT

This tests a different part of the engine: joins plus aggregation plus top-N processing.

3. Ranking within each region

The third query performs an aggregation followed by a ranking window:

Join
  ↓
Aggregate
  ↓
Window / Rank
  ↓
Filter rank <= 3

This is useful because it combines several operators and exercises the ability to keep a larger section of the plan columnar.

The most useful part: same code, two execution modes

The example deliberately runs each query twice with just a config change. It also inspects the final physical plan to show which operators were accelerated.

This makes the experiment much more interesting than simply comparing two unrelated implementations.

We are comparing:

Same PySpark code
Same Spark session
Same data
Same query
        │
        ├── Spark execution
        │
        └── VecRuntime execution

What does the example show?

On my Apple M3 Pro system, using Spark 4.1.3 and 20 million rows, the repository reports the following indicative results ( relative speed up):

• 1.19× for the aggregation query
• 1.24× for the join/top-N query
• 1.17× for the ranking query

These are intentionally presented as indicative rather than universal benchmarks. The example is running locally on one machine, the machine was doing other work, and the data size is relatively small.

The more interesting observation is what happens without the columnar shuffle. For this workload, the README notes that stage boundaries can spend enough time converting between columnar batches and rows that much of the operator-level gain disappears.

Faster operators alone are not enough. The execution pipeline has to remain columnar.

The kernels are where the fun starts

The repository also contains lower-level JMH benchmarks for the Java kernels. These are kernel-level numbers, not end-to-end Spark query speedups.

That distinction matters enormously. A kernel can be 8× faster while the overall query only becomes 1.2× faster because the application also spends time scanning Parquet, scheduling tasks, shuffling, materializing results, performing unsupported expressions and so on.

This is precisely why the columnar pipeline matters.

Aggregation is more subtle

Vectorization is not automatically beneficial everywhere. Grouped aggregation is a good example.

A naive implementation might perform a SIMD reduction separately for every group. That sounds attractive.

But imagine four groups and a batch of thousands of rows. You don’t want to scan the entire batch four times. The implementation therefore uses a scatter-oriented approach where rows update grouped accumulators.

SIMD alone doesn’t make an algorithm efficient.

The algorithm still has to respect memory access patterns, cache behavior and data dependencies.

Floating-point correctness is another trade-off

There is another interesting issue with vectorized aggregation: floating-point associativity. Spark’s normal aggregation order can produce a particular rounding result. A vectorized reduction may use multiple partial sums and therefore perform additions in a different order.

But floating-point arithmetic can produce slightly different results. VecRuntime therefore has a strict floating-point mode. With strict mode enabled, it can reproduce Spark’s rounding semantics rather than simply maximizing SIMD throughput.

This is a useful example of a principle that gets overlooked in vectorized execution projects:

Performance is only useful when the semantics are correct.

And this is still Spark

One of the architectural decisions I particularly like is the fallback behavior. VecRuntime does not try to implement every Spark expression and operator before users can try it. Instead, it can partially accelerate a plan. For example:

VectorFilter
     ↓
VectorProject
     ↓
SparkPythonUDF
     ↓
SparkWrite

The vectorized operators run where possible, and Spark handles the rest. The README lists unsupported or non-vectorized areas (yet). This makes the approach much closer to an accelerator layer than a replacement Spark engine.

There is also a Spark UI

One of the nicest features of the example is the Vector Acceleration tab in the Spark UI.

With:

uv run example.py --keep-alive

The example keeps the Spark session alive and exposes the Spark UI. The custom tab visualizes the execution plan and shows which operators were executed by VecRuntime and which remained on Spark.

That is especially useful for debugging partial acceleration. Instead of asking:

“Why didn’t this query speed up?”

you can ask:

“Which operator prevented the pipeline from remaining vectorized?”

That is a much more actionable question.

Looking beyond a local laptop

The local PySpark example is intentionally simple. The main project has also been tested on larger workloads, including TPC-H and TPC-DS.

The current project documentation reports a 1 TB TPC-DS comparison on EKS using eight 13-core executors.

In that configuration, the documented current as publication date totals are:

They demonstrate that a JVM-native execution path can be pushed far beyond a small local experiment and can execute a substantial analytical workload.

The project reports that accelerated 82 of the 103 TPC-DS queries in that campaign.

JVM-native does not mean anti-native

An interesting part of the design is that the project is designed to coexist with Apache DataFusion Comet.

A potential pipeline looks like:

Native Parquet decoding
        ↓
Arrow / columnar representation
        ↓
VecRuntime
        ↓
Comet or other native component

The common currency is the columnar representation. That opens up a more interesting architecture than the usual:

JVM engine vs native engine

Instead, we can think about:

Different execution components sharing the same columnar memory representation.

That is a much more composable model.

Final thoughts

There is a lot of attention on native execution engines for Spark. And for good reason. But the Java ecosystem has an interesting capability that is still relatively under-explored for analytical execution: the Java Vector API.

VecRuntime is an experiment around that idea. The goal is not to replace Spark. It is to ask whether Spark’s familiar execution model can be pushed further by:

Not necessarily need to leaving the JVM to get a vectorized engine.

Resources

• VecRuntime: https://github.com/vecruntime/vecruntime
• PySpark example: https://github.com/vecruntime/pyspark-example
• Java Vector API: https://openjdk.org/jeps/338
• Apache DataFusion Comet: https://github.com/apache/datafusion-comet
• Apache Arrow: https://arrow.apache.org/

The itch

If you’ve used DuckDB-Wasm: the WebAssembly build of DuckDB that runs entirely in the browser and want to use with AWS platform you may have hit this:

CREATE SECRET s3 (TYPE S3, PROVIDER CREDENTIAL_CHAIN);

Extension Autoloading Error: An error occurred while trying to automatically
install the required extension 'aws':
Extension https://extensions.duckdb.org/v1.1.3/wasm_eh/aws.duckdb_extension.wasm
is not available

The aws extension simply isn't distributed for the wasm platforms. There's a long-standing issue about it (duckdb-wasm#1919), and the extension's CI config is refreshingly blunt about why:

exclude_archs: 'wasm_mvp;wasm_eh;wasm_threads;...' # Doesn't work anyway

“Doesn’t work anyway” is a challenge if I’ve ever read one. I wanted to know two things: does it actually not compile, and if we fix that, is it actually useless in a browser?

Spoiler: it didn’t compile, the fixes are small, and it’s much less useless than expected.

Blocker #1: the AWS event loop that wasn’t there

The aws extension pulls in the full native AWS stack via vcpkg: aws-sdk-cpp, the AWS Common Runtime (CRT) libraries, OpenSSL, curl, s2n, zlib. My assumption was that this pile of native networking code would explode immediately on wasm32-emscripten.

It mostly… didn’t. OpenSSL built. curl built. s2n built. Almost every aws-c-* library built. Exactly one dependency failed:

CMake Error at CMakeLists.txt:170 (message):
  Event Loop is not setup on the platform.

aws-c-io — the CRT's socket and event-loop layer — selects an implementation per platform: IOCP on Windows, epoll on Linux, kqueue on BSD, dispatch queues on Apple. Emscripten matches none of the branches, and the build hard-fails.

Here’s the thing though: the extension never needs an event loop. Its code paths go through the SDK’s classic curl-style HTTP client, not the CRT’s async engine. And aws-c-io's own source already contains graceful runtime stubs for unsupported loop types — the only hard failures were the CMake check and one #error in the default-type lookup.

So the fix is a vcpkg overlay port with a small patch: add an Emscripten branch that compiles the POSIX sources (sockets, pipes, DNS — these build cleanly against Emscripten’s musl libc) and ships no event loop. Creating one at runtime returns AWS_ERROR_PLATFORM_NOT_SUPPORTED instead of failing the build:

elseif (CMAKE_SYSTEM_NAME STREQUAL "Emscripten")
    file(GLOB AWS_IO_OS_SRC "source/posix/*.c")
    list(APPEND EVENT_LOOP_DEFINES "EMSCRIPTEN_NOOP")
    set(USE_S2N ON)

With the overlay in place, the entire dependency tree — aws-sdk-cpp included — built for wasm32-emscripten. First surprise of the project.

Blocker #2: the 420 KB extension that should have been 12 MB

The build produced aws.duckdb_extension.wasm… at 420 KB. For something that statically links the AWS SDK, that's not lean, that's empty. Inspecting the module confirmed it: thousands of Aws::* symbols were unresolved imports, expected to come from the main DuckDB module at load time. They never would — DuckDB-Wasm doesn't contain the AWS SDK.

The cause is a quirk of how DuckDB builds wasm extensions. A loadable wasm extension isn’t linked by CMake’s normal machinery; it’s produced by a post-build re-link:

emcc <extension-archive> -o <name>.duckdb_extension.wasm -O3 -sSIDE_MODULE=2 ${LINKED_LIBS}

target_link_libraries is ignored for this step entirely. Any third-party static libraries must be passed explicitly through a LINKED_LIBS parameter in the extension's extension_config.cmake — which the aws extension never sets, because nobody had ever linked it for wasm.

The fix is boring and beautiful — list the vcpkg-built archives:

duckdb_extension_load(aws
    SOURCE_DIR ${CMAKE_CURRENT_LIST_DIR}
    LOAD_TESTS
    LINKED_LIBS "../../vcpkg_installed/wasm32-emscripten/lib/libaws-cpp-sdk-core.a ..."
)

Result: a 9 MB side module where every AWS symbol resolves internally, and the only remaining imports are libc/libc++ functions the DuckDB-Wasm main module genuinely provides.

Blocker #3: four missing byte-swap functions

Almost. The main module doesn’t export ntohs, htons, ntohl, htonl. Twenty-eight lines of C shims later (wasm32 is little-endian; these are plain byte swaps), the module was self-contained.

Does it load?

duckdb-wasm version: v1.5.5
LOAD aws: ✅ OK
duckdb_extensions(): [{"extension_name":"aws","loaded":true}]

Yes, in Node and in Chrome, against a stock @duckdb/duckdb-wasm build. One gotcha worth knowing: DuckDB refuses extensions built against a different DuckDB version, so the extension must be compiled against exactly the version your duckdb-wasm package wraps.

And the “doesn’t work anyway” claim? It’s half right. The extension’s reason for existing is the AWS credential chain: environment variables, ~/.aws config files, SSO caches, IMDS, STS. A browser tab has none of those. CHAIN 'env' and CHAIN 'config' fail cleanly with "no credentials found". The default chain goes further and tries to open a real socket for the IMDS probe , which dies on a missing _ntohs in duckdb-wasm's own Emscripten JS glue (a fixable bug on their side, and sockets wouldn't work in a browser regardless).

So: compiles, loads, runs, and fails exactly where a browser sandbox says it must. But while testing that, I found the actual treasure.

The plot twist: AWS’s endpoints are browser-friendly

To exercise the extension I wanted a real workload: Amazon S3 Tables, AWS’s managed Apache Iceberg service, which DuckDB’s iceberg extension can attach as a catalog without need for aws one:

CREATE SECRET s3s (TYPE S3, KEY_ID '…', SECRET '…', SESSION_TOKEN '…', REGION 'us-east-1');
ATTACH 'arn:aws:s3tables:…:bucket/demo' AS s3t (TYPE iceberg, ENDPOINT_TYPE s3_tables);

CREATE SCHEMA IF NOT EXISTS s3t.webdemo;
CREATE TABLE s3t.webdemo.sensor_readings (sensor_id VARCHAR, ts TIMESTAMP, temperature DOUBLE, humidity DOUBLE);
INSERT INTO s3t.webdemo.sensor_readings SELECT … FROM range(100);
SELECT sensor_id, count(*), round(avg(temperature),1)
FROM s3t.webdemo.sensor_readings GROUP BY sensor_id;Every one of those statements executes in the browser. The CREATE TABLE and INSERT are real Iceberg commits — catalog metadata via the S3 Tables REST API, parquet files PUT to the bucket, all signed client-side. I verified from native DuckDB afterwards: the 100 rows written by the browser are really there.

The demo

I wrapped this in a small React app (repo link) with three steps:

1. Credentials: paste a temporary key/secret/session token and a table bucket ARN. DuckDB-Wasm boots, loads the locally built aws extension plus the official httpfs and iceberg wasm builds, creates the secret, attaches the catalog. Note that the demo does not need aws extension as it is already supported on latest icebergextensions.

More on this setup here.

A quick-startup guide to connect DuckDB Shell to Amazon S3 Table Bucket

1. Generate: synthesizes IoT sensor readings and commits them to a new Iceberg table in your bucket.
2. Query: preset and free-form SQL against the live table, with timings.

The repo ships the compiled aws.duckdb_extension.wasm and the full patch set with build instructions, so you can reproduce the binary from source.

What should happen upstream

I’ve packaged the three fixes as a draft PR against duckdb-aws that I will try to upstream soon, with the aws-c-io overlay port needing a small companion change in the extension CI tooling. Honest caveats included: shipping the extension for wasm is only step one.

The genuinely interesting follow-up belongs in duckdb-wasm itself: a JS-side credential callback. Browsers do have credential sources: OAuth flows, Cognito, STS AssumeRoleWithWebIdentity via fetch, they're just JavaScript-shaped, not file-shaped.

Let the host application hand credentials to the aws extension and the browser story becomes complete: sign in with your IdP, query your lakehouse, close the tab. Will try to work on this next.

Takeaways

• “Not supported” often decomposes into two or three small, fixable reasons. Here: one missing CMake branch, one missing link flag, four byte-swap functions.
• DuckDB’s wasm extension link step ignores target_link_libraries, if you're building an extension with native dependencies for wasm, you need LINKED_LIBS.
• AWS quietly made S3 Tables browser-accessible: permissive CORS on both the Iceberg REST catalog and the data plane. Combined with client-side SigV4 in DuckDB’s iceberg extension, serverless-serverless analytics on Iceberg is already real.

Built with DuckDB v1.5.5, duckdb-wasm, emsdk 3.1.71 and vcpkg. All code and patches are developed with help of Fable: repo link.

This also means that you can use the official Spark dbt connector now.

This post builds a small but complete example: a dbt project running entirely from a laptop, materialising Apache Iceberg v3 tables in the Glue Data Catalog, using the VARIANT data type to absorb semi-structured IoT payloads whose shape changes per firmware version.

The code can be found here:

GitHub - Neuw84/dbt-glue6: Demo of Spark Connect with dbt on Glue 6

Why Spark Connect changes the integration story

Before Spark Connect, “run dbt on Glue” meant one of two things. Either you submitted code to Glue and let something on the cluster interpret it, or you used an adapter that reimplemented dbt’s execution against a Glue-specific API. Both approaches put a translation layer between dbt and Spark, and translation layers drift.

Spark Connect inverts the arrangement. The client holds a SparkSession object that is a thin proxy: it serialises unresolved logical plans over gRPC, the remote driver analyses and executes them, and results come back as Arrow batches. Nothing about the client needs to know it is talking to Glue.

So the entire dbt integration is one environment variable:

env["SPARK_REMOTE"] = remote_url          # sc://...
subprocess.run([sys.executable, "-m", "dbt.cli.main", "build"], env=env)

with a stock profile:

dbt_glue_demo:
  target: dev
  outputs:
    dev:
      type: spark
      method: session      # dbt asks PySpark for a SparkSession...
      host: localhost      # ...unused, but dbt-spark requires the key
      schema: dbt_glue_demo
      threads: 4

dbt-spark's session method calls SparkSession.builder.getOrCreate(). PySpark sees SPARK_REMOTE and connects to Glue. That is the whole seam — dbt-core 1.12.3 and dbt-spark 1.11.0 straight from PyPI, no patches.

Remember that you can also use the Glue dbt adapter if that suits you.

The three-call handshake

Getting an sc:// URL out of Glue takes three API calls. There is no long-lived endpoint to hard-code.

1. Create the session. SessionType="SPARK_CONNECT" is the switch:

glue.create_session(
    Id="dbt-glue-session",
    Role=session_role_arn,
    Command={"Name": "glueetl", "PythonVersion": "3"},
    GlueVersion="6.0",
    SessionType="SPARK_CONNECT",
    WorkerType="G.1X",
    NumberOfWorkers=2,
    IdleTimeout=30,          # minutes of inactivity before Glue stops it
    Timeout=120,             # hard lifetime ceiling
    DefaultArguments=default_arguments,
)

2. Poll until READY. get_session reports PROVISIONING then READY. In testing this took about 30 seconds for a 2× G.1X session. Treat FAILED, STOPPED, TIMEOUT and STOPPING as terminal and stop waiting.

3. Fetch the endpoint. This is where the connection string comes from:

endpoint = glue.get_session_endpoint(SessionId=session_id)["SparkConnect"]
token = urllib.parse.quote(endpoint["AuthToken"], safe="")
remote = f"{endpoint['Url']}:443/;use_ssl=true;x-aws-proxy-auth={token}"

Three details will bite you if you skip them:

• Append :443/ yourself. Url comes back without a port or trailing slash; PySpark's Spark Connect URL parser needs both.
• URL-encode the token. It contains characters that would otherwise terminate the ;-delimited parameter list. safe="" matters.
• The token lives 30 minutes. AuthTokenExpirationTime says exactly when. The runner therefore mints a fresh token immediately before launching dbt, so the full window is available to the run.

That 30-minute ceiling is a real constraint, not a formality. dbt holds one SparkSession for the lifetime of the invocation, so a token refreshed mid-run cannot be handed to it. Runs longer than 30 minutes need to be split, or driven by a scheduler that re-establishes the session per batch of models.

Configuring Iceberg once, for every model

The interesting part of DefaultArguments is the Iceberg catalog. Glue's convention here is unusual: all Spark settings go under a single --conf key, joined by the literal string " --conf ".

default_arguments = {
  "--datalake-formats": "iceberg",
  "--enable-spark-live-ui": "true",
  "--conf": " --conf ".join([
    "spark.sql.extensions=org.apache.iceberg.spark.extensions.IcebergSparkSessionExtensions",
    "spark.sql.catalog.glue_catalog=org.apache.iceberg.spark.SparkCatalog",
    "spark.sql.catalog.glue_catalog.catalog-impl=org.apache.iceberg.aws.glue.GlueCatalog",
    "spark.sql.catalog.glue_catalog.io-impl=org.apache.iceberg.aws.s3.S3FileIO",
    f"spark.sql.catalog.glue_catalog.warehouse={warehouse}",
    "spark.sql.catalog.glue_catalog.table-default.format-version=3",
    "spark.sql.defaultCatalog=glue_catalog",
  ]),
}

The line that does the heavy lifting for this post:

spark.sql.catalog.glue_catalog.table-default.format-version=3

table-default.* applies to every table the catalog creates. Set it once and every dbt model comes out as Iceberg v3 with no per-model TBLPROPERTIES. Verified across all four relations in the project — seed, staging, and both marts — each reports format-version=3.

Doing it at the catalog level rather than per model is deliberate. v3 is a one-way door: a table can be upgraded from v2 to v3, but not downgraded. Making it the catalog default means you decide once, explicitly, in infrastructure code, instead of hoping seventeen models all remembered the same property.

Setting spark.sql.defaultCatalog=glue_catalog lets dbt's generated SQL use bare database.table names.

VARIANT: one column, many shapes

The scenario is IoT device telemetry. Each device sends JSON, and the shape depends on what the device measures and which firmware it runs:

{"temp_c": 21.5, "battery": {"pct": 88, "charging": false}, "tags": ["indoor", "floor-1"]}
{"temp_c": 30.1, "humidity_pct": 55, "fw": "2.3.1"}
{"lat": 40.4168, "lon": -3.7038, "speed_kmh": 12.4}
{"error": {"code": 503, "msg": "sensor fault (recalibrated)"}, "battery": {"pct": 85, "charging": true}}
{"temp_c": 19.8, "humidity_pct": 63, "air_quality": {"pm25": 8, "co2_ppm": 480}, "fw": "4.0.0"}

The seed’s seventeen rows produce around ten distinct payload shapes. The classic options are all unpleasant: a sparse wide table that grows a column per firmware release, a STRING column re-parsed on every query, or a schema migration each time a device team ships an update.

VARIANT is a fourth option. The staging model promotes the raw JSON text into it:

select
    cast(event_id as bigint)       as event_id,
    device_id,
    event_type,
    cast(ingested_at as timestamp) as ingested_at,
    parse_json(payload_json)       as payload
from {{ ref('raw_device_events') }}

parse_json is not a cosmetic cast. The value is stored in Iceberg's binary VARIANT encoding, and Glue shreds frequently-seen fields into hidden Parquet sub-columns — so filters and projections get column statistics and file pruning instead of a full-text re-parse per row.

The mart then lifts the fields it cares about into typed columns:

select
    event_id,
    device_id,
    event_type,
    ingested_at,

    payload:temp_c::double            as temp_c,
    payload:humidity_pct::int         as humidity_pct,
    payload:speed_kmh::double         as speed_kmh,

    payload:battery.pct::int          as battery_pct,     -- nested object
    payload:battery.charging::boolean as battery_charging,

    payload:error.code::int           as error_code,      -- only on error events
    payload:error.msg::string         as error_message,

    payload:tags[0]::string           as primary_tag,     -- array element

    payload:fw::string                as firmware,
    schema_of_variant(payload)        as payload_shape    -- handy for debugging
from {{ ref('stg_device_events') }}

payload:a.b navigates; ::type casts. A field a given firmware never sends comes back NULL. No CASE pyramid, no coalesce chain, no migration.

Aggregation works the same way, straight out of the VARIANT with no intermediate flattening:

select
    device_id,
    count(*)                              as event_count,
    max(payload:fw::string)               as latest_firmware,
    round(avg(payload:temp_c::double), 2) as avg_temp_c,
    min(payload:battery.pct::int)         as min_battery_pct,
    count(payload:error.code)             as error_count
from {{ ref('stg_device_events') }}
group by device_id

avg silently skips rows whose payload lacks the field, which is the desired behaviour: GPS events should not drag the temperature average toward zero.

Also verified working: variant_get(payload, '$.tags', 'array<string>') for extracting a whole array with an explicit type, and schema_of_variant for inspecting a row's inferred shape.

The caveat: number inference is per row

This one cost real debugging time. schema_of_variant reports the schema Spark infers for that individual row, and numeric inference is value-dependent:

JSON literal Inferred type21.5DECIMAL(3,1)3BIGINT1.0e3DOUBLE"x"STRING

So the same logical field differs between rows. In this dataset speed_kmh came back as DECIMAL(3,1) for 12.4 but DECIMAL(1,0) for 0.0; lat was DECIMAL(6,4) in one row and DECIMAL(5,3) in another.

The practical rules:

1. Always cast explicitly when projecting: payload:speed_kmh::double, never the bare extraction. The cast is what gives you a stable column type.
2. Never compare schema_of_variant output across rows to decide whether two payloads "match". It is a debugging aid, not a schema registry.

Incremental MERGE, and why the write mode matters

fct_device_telemetry is incremental, merged on event_id:

{{ config(
    materialized='incremental',
    incremental_strategy='merge',
    unique_key='event_id',
    tblproperties={
        'write.merge.mode': 'merge-on-read',
        'write.update.mode': 'merge-on-read',
        'write.delete.mode': 'merge-on-read',
    }
) }}

To exercise it, the seed was extended after a first full load: a correction to an already-loaded event (same event_id, changed message) plus new events, one of which carried a brand-new nested object — air_quality — that no earlier row contained.

Both landed correctly. The correction updated in place, the new event inserted, and air_quality was absorbed with zero schema change — 17 rows, 17 distinct event_ids. A genuinely new nested structure required no DDL, because as far as the table is concerned nothing about the column changed.

Remember that Iceberg defaults to copy-on-write. Copy on write in this use case is like 8x times slower.

Running it

Infrastructure is deliberately small: S3 warehouse bucket, Glue Catalog database, the session execution role, and a policy for your local identity. Eleven resources.

cd terraform
terraform init
terraform apply

terraform apply also writes generated/glue_session.json, which the runner reads. Infrastructure and runner therefore cannot drift: change the worker type in Terraform and the next dbt run picks it up.

cd runner
uv run python run_dbt.py build

uv handles the local environment. The client pins pyspark==4.1.1 exactly, to match the server's 4.1.1-amzn-0 — Spark Connect tolerates some skew, but matching removes a variable. Local Python is pinned to 3.12 rather than the server's 3.13, purely for dbt dependency wheel availability; the client and server Python versions are independent.

Result:

Completed successfully
Done. PASS=17 WARN=0 ERROR=0 SKIP=0 TOTAL=17

Four Iceberg v3 tables and thirteen dbt tests, from a laptop, against a managed Spark cluster.

The session is left warm between invocations — reattaching to a READY session skips the ~30 s provisioning wait. It auto-stops after IdleTimeout minutes, or immediately with --stop-session.

Things to know before you build on this

No Lake Formation fine-grained access control. Spark Connect sessions do not support LF-FGAC. If row/column-level Lake Formation policies are a requirement, this route is not available to you today — use a conventional Glue job.

Statement APIs do not apply. RunStatement / GetStatement are for notebook-style sessions. A Spark Connect session is driven entirely through the gRPC endpoint; the statement APIs are not part of the picture.

Private networking uses a dedicated endpoint. For a session reachable from a VPC without traversing the internet, the interface endpoint is com.amazonaws.{region}.glue.sessions — distinct from the regular com.amazonaws.{region}.glue endpoint.

The 30-minute token is the main architectural constraint. It bounds a single dbt invocation. For longer pipelines, split the work across invocations, each minting its own token, rather than trying to refresh inside a held session.

One enableHiveSupport warning is expected. dbt-spark's session method calls it, and Spark Connect rejects it with CANNOT_MODIFY_CONFIG. It is harmless — the Iceberg catalog is already the session default.

Wrapping up

Spark Connect on Glue 6.0 removes the adapter problem rather than solving it. dbt does not know it is talking to Glue; Glue does not know it is running dbt. The contract is a gRPC endpoint and an environment variable, and everything on either side stays stock.

What that buys, concretely: dbt build from a laptop against managed Spark, in Iceberg v3 tables, with VARIANT columns that absorb per-firmware payload variation and deletion vectors that made incremental merges 8.6× faster against copy on write mode.

References

• AWS Glue interactive sessions with Spark Connect
• Apache Iceberg table spec v3
• Spark Connect overview
• Spark VARIANT data type
• dbt-spark adapter

Two years ago I published a benchmark on streaming workloads in a GitHub repository.

GitHub - aws-samples/iceberg-streaming-examples: This repo contains examples of high throughput ingestion using Apache Spark and Apache Iceberg. These examples cover IoT and CDC scenarios using best practices. The code can be deployed into any Spark compatible engine like Amazon EMR Serverless or AWS Glue. A fully local developer environment is also provided.

I also used the code to push as much as 2 million events per/second while being able to ingest them with MERGE with less than one hour of latency using V2 tables. Note: for that 2 million/sec I used a huge EMR Serverless cluster.

With all the ecosystem evolving to Iceberg Specification V3 I wanted to see how much I could push the numbers now with V3.

Apache Iceberg format version 3 ships two headline features for row-level workloads:

• Deletion Vectors (DVs), one compact bitmap per data file instead of piles of position-delete files.
• Row Lineage, which gives every row a stable _row_id. Both are exactly what you want under a high-throughput CDC MERGE.

I upgraded my streaming benchmark from v2 to v3 and my executors started dying like clockwork while on v2 they were completely stable, DVs were the obvious suspect.

The setup

Kafka → Spark Structured Streaming → Iceberg, a DMS-style CDC feed doing a hot-key MERGE at ~90,000 rows/second into a bucket(64) merge-on-read mirror table.

Spark 4.0.2 on EKS, Iceberg 1.11, six executors with 8 cores, 32 GB of heap and 8 GB of overhead each and 40 GiB pod limits.

On format version 2, this runs all day. On format version 3, every run followed the same script:

• t+75 minutes: the first wave of executors is OOM-killed by the kernel.
• Spark spins up replacements. They die within 5–7 minutes.
• The driver follows shortly after.

Same code, same knobs, same feed. The only variable was format-version=3. I ran the v2 control for 6 hours and 46 minutes with a memory line flat as glass to be sure.

So: something v3-specific, growing over time, unbounded was killing the benchmark.

Act 1: chasing the obvious suspect

v3’s MERGE path is structurally different. With DVs, a row-level MERGE becomes a read-modify-write: previous deletion vectors are loaded, merged with new deletes, and rewritten every batch. Reading the Spark Iceberg write path (SparkPositionDeltaWrite), two genuinely suspicious terms fell out of code review with Fable ( and careful after double from me).

1. Driver side: every batch broadcasts a map of every data file that has deletes → its delete files, with full metadata (partition data, column stats and bounds included). On a growing table with a hot-key MERGE this converges to “all files in the table” and grows monotonically.
2. Executor side: BaseDVFileWriter buffers one in-memory bitmap per touched data file for the whole task and only merges each file's previous DV at close() — all of it resident simultaneously. Peak state is O(files touched by the task).

Both are real. I fixed both with Fable’s help:

• Slim-broadcast patch that strips the metadata executors never read (measurably 1.4–1.8× smaller serialized broadcasts).
• Clustered DV writer that exploits Spark’s sorted-input mode to hold exactly one file’s bitmap at a time.

• Spark: Broadcast slim rewritable delete metadata in position delta wr... by Neuw84 · Pull Request #17281 · apache/iceberg
• Core, Spark: Add clustered DV writer for ordered position delta writes by Neuw84 · Pull Request #17287 · apache/iceberg

Then, I did the honest thing and ran a three-arm A/B at identical sizing, same feed, one variable per arm.

Ouch. The sorted-input mode (fanout=false) bought fifteen minutes. Our shiny bounded-memory writer bought three more minutes (inside run noise). Executor working sets in both arms climbed the same curve, ~50-80 MiB/min, straight into the 40 GiB pod limit.

Meanwhile the most telling metric was quietly sitting in the Spark REST API the whole time: peak JVM heap was frozen at ~28–29 GB in every arm, for hours, to the megabyte.

The heap wasn’t growing. The pod was. Whatever was eating memory lived outside the JVM heap, and I had just spent 3 days optimizing heap-side data structures. Nervertheless, this is an architectural improvement, O(1 file) instead of O(files touched), that matters for file-heavy tables and tight heaps.

Note on AI : without Fable´s help this could be a work of weeks. Fable helped me out to run it on top of EKS in distributed manner ( I discovered the bug on EMR Serverless but needed OSS confirmation) for quicker A/B testing, also developed the monitoring scripts that allowed to conclude that on-heap code wasn´t the problem. If you look at the code from those pull requests you can see that there isn´t much but they are a worthy add and the code is clean ( note that I have a long time Java background I and I am already familiar with Spark and Iceberg code that helps on guidance).

Act 2: stop guessing, measure the off-heap

This was the turning point, and it was a methodology decision, not a code insight: no more hypotheses until we could name every byte in the container. The plan:

1. Ask the kernel first. Before killing a doomed run, we exec’d into an executor pod sitting 500 MiB from its limit and read cgroup v2’s memory.stat:

anon 42514927616     # 42.5 GB of anonymous memory
file 285040640       # 285 MB of page cache

That single read eliminated half the hypothesis space. This was not page cache from the Puffin/Parquet churn (which the kernel could reclaim under pressure): it was real, allocated, unreclaimable memory. The JVM process RSS was 41.8 GB against a 32 GB heap: ~10 GB of native memory, and growing.

2. Make the JVM confess. Note that Java have very powerful configuration options, I relaunched with -XX:NativeMemoryTracking=summary and a monitor that sampled jcmd VM.native_memory plus memory.stat from every executor every two minutes. The time series was unambiguous:

workingSet   anon    NMT "Other" (direct buffers)
t+05min       33.7 GiB   33.5      157 MB
t+20min       35.0       34.7      337 MB
t+40min       36.1       35.8    1,229 MB
t+60min       37.4       37.1    2,638 MB
t+80min       38.8       38.4    4,010 MB
t+87min       39.9       39.8    5,446 MB   ← pod killed

Every other NMT category , heap, GC, metaspace, code cache, threads was flat for the entire run.

The killer was direct ByteBuffers, accumulating at ~60 MB/min and accelerating with table size. But, what was causing this?

Note on AI: as long time Java developer ( used to do profiling back to Netbeans 5 era), I knew how powerful Java monitor is, however, Fable and Kiro helped me a lot of doing the setup and running the experiments that would cost me days ( getting configs, automating EKS A/B runs, monitoring scripts).

3. Set a trap. Direct buffers are only released when their Java wrapper objects get garbage-collected, and a 32 GB heap that never fills rarely runs old-gen collections. So were the buffers garbage (waiting for a GC that never came, a config fix) or leaked (still reachable bug)? One flag distinguishes them: -XX:MaxDirectMemorySize=4g.

When the JDK hits that cap it forces a full GC and retries before throwing. If the run stabilizes into a sawtooth: garbage. If it throws: leak, and the stack trace names the allocator. Seventy-eight minutes later, all six executors answered within seconds of each other:

java.lang.OutOfMemoryError: Cannot reserve 4194304 bytes of direct buffer memory
    (allocated: 4292201200, limit: 4294967296)
  at java.nio.ByteBuffer.allocateDirect(ByteBuffer.java:332)
  at ...shaded.io.netty.buffer.PoolArena$DirectArena.newChunk(PoolArena.java:737)
  at ...shaded.org.apache.arrow.vector.BaseFixedWidthVector.allocateNew(...)
  at org.apache.iceberg.arrow.vectorized.VectorizedArrowReader.allocateBigIntVector(...)
  at org.apache.iceberg.arrow.vectorized.VectorizedArrowReader$RowIdVectorReader.read(...)
  at org.apache.iceberg.spark.data.vectorized.ColumnarBatchReader.read(...)

A true leak, the forced GC could not reclaim the buffers — and a named culprit that made everything click at once: RowIdVectorReader. The reader that materializes the v3 row lineage _row_id metadata column.

Not the deletion vectors. Not the MERGE writer. The read side of v3’s other new feature, a code path that v2 tables never instantiate. Which is why the v2 control ran flat for seven hours: it was never about how we wrote deletes at all.

Note on AI: note that I already knew that I could cap to see what was going on but with Kiro and Fable´s help I run these experiments in an afternoon!

Act 3: the bug, in fifteen lines

The offending pattern (Iceberg’s iceberg-arrow module, paraphrased):

@Override
public VectorHolder read(VectorHolder reuse, int numValsToRead) {
  ...
  // fresh direct-memory vector allocated EVERY batch; `reuse` is ignored
  BigIntVector rowIds = allocateBigIntVector(ROW_ID_ARROW_FIELD, numValsToRead);
  ...
  return VectorHolder.vectorHolder(rowIds, MetadataColumns.ROW_ID, nulls);
}

@Override
public void close() {
  // don't close result vectors as they are not owned by readers
}

The reader allocates a fresh Arrow BigIntVector (backed by netty pooled direct memory, 4 MiB arena chunks, exactly the allocation size in the OOM) for every column batch, ignores the reuse parameter that exists for precisely this purpose, keeps no reference to what it handed out, and closes nothing.

Nothing downstream closes the holder's vector either. One leaked vector per batch, per lineage column, forever. Its sibling LastUpdatedSeqVectorReader (_last_updated_sequence_number) has the identical pattern.

The fix is exactly what the base class already does for regular columns: the reader owns its result vector, allocate lazily, reuse across batches while capacity suffices, release in close().

Plus regression tests that assert the Arrow root allocator returns to its baseline after close, tests that fail against the unfixed code.

Arrow: Fix direct memory leak in row lineage vectorized readers by Neuw84 · Pull Request #17296 · apache/iceberg

Note on AI: Kiro with Fable implemented the fix with some tests, note that the code is quite simple to understand if you are familiar with Java code. That is a plus in times where review is the bottleneck for AI generated code. Note that implementing this, along test would take it me like two or three days of work.

Act 4: proof

Same cluster, same feed, same sizing, no tuning flags ,only the reader fix.

Thirteen hours of dead-flat memory against a 92-minute baseline death. v3 streaming MERGE at 90k rows/s is simply stable now,no extra overhead, no direct-memory cap, no compaction crutch.

You still want compaction for read amplification as batch times grow as uncompacted DVs pile up, but that’s now a throughput concern.

What working with an AI agent was actually like

I did this investigation pair-programming with an AI coding agent (Kiro) and with Fable, and it’s worth being precise about the division of labor, because it wasn’t “AI finds bug”.

What the agent did well: volume and velocity. It read the unfamiliar Iceberg write path and produced the two initial hypotheses in minutes. It wrote every monitoring script (pod-state watchers, NMT samplers, log collectors with rotation-proof capture), drove the build-image-deploy-run loop dozens of times, kept multi-hour experiments under observation, extracted trend tables from raw logs, wrote the patches, the regression tests, and the microbenchmark. When the OOM stacks landed at 2 a.m. cluster time, the collector it had written twenty minutes earlier had already saved them with full context. It was also honest under pressure: when the three-arm A/B showed our own writer bought nothing, the summary said so in the first sentence.

What the human (me) did: strategy and skepticism. Deciding that a v2 control run mattered before any code changed. Refusing to accept “memory grows” without an anon-vs-file-cache split. Insisting we instrument off-heap when the heap numbers looked too clean. Choosing the direct-memory cap as a trap rather than a mitigation. And repeatedly asking the uncomfortable question:“but did our fix actually change anything?” that turned two plausible optimizations into footnotes and kept the investigation moving toward the real bug.

The agent multiplied the experiments per day; the human decided which experiments deserved to exist. Neither half finds this bug alone in a week ( as personal note, this would took at least 2 weeks of full time job whereas I could manage to discover and fix it in 3 days.

The runbook, if your executors are dying

1. Run the control. One variable. If v2-vs-v3 (or old-vs-new anything) flips the behavior, you’ve halved the search space before touching a profiler.
2. Watch peak JVM heap vs container working set. If heap is flat and the pod climbs, stop optimizing your data structures, the answer is off-heap.
3. memory.stat before the pod dies. anon vs file is one kubectl exec and instantly separates "real allocation" from "reclaimable cache".
4. NMT is cheap. -XX:NativeMemoryTracking=summary plus jcmd sampling attributes the anon growth: heap, GC, threads, code or Other, which means direct buffers.
5. Cap direct memory to force a confession. -XX:MaxDirectMemorySize makes the JDK full-GC-and-retry before throwing. Sawtooth = garbage, config problem. OOME = leak, and the stack trace is your suspect, named and addressed.
6. Validate with the same telemetry that caught the bug. Flat lines for thirteen hours are worth more than any argument.

Everything is on its way to be upstreamed, hopefully soon :).

V3’s features are worth it. Now they’re also stable at 90k rows a second.

The investigation, experiments and patches described here were done in collaboration with an AI coding agent; all findings were verified against live cluster telemetry, and all code was reviewed and tested before submission upstream.

That story is changing. With Apache Spark Connect now available on Amazon EMR Serverless, and Amazon S3 Tables giving you managed Apache Iceberg with zero table maintenance, you can point dbt at a serverless Spark endpoint from your laptop and transform lakehouse data as if it were any other warehouse. No cluster to size, no cluster to forget to turn off.

This post walks through a complete, Terraform-managed example that does exactly that. The same dbt code runs from your laptop for development and on Amazon MWAA (managed Airflow) for scheduling. Everything is serverless, and the whole thing tears down with a single terraform destroy.

The repo where the code lives is:

GitHub - aws-samples/sample-dbt-spark-connect-emr-serverless-s3tables

I’ll focus on the parts that were genuinely interesting to get right, because a few of them are not in the happy-path docs yet.

This blog post is based from some previous ideas from some colleages:

https://aws.amazon.com/blogs/big-data/serverless-analytics-pipelines-using-the-apache-spark-engine-in-amazon-athena/

https://aws.amazon.com/blogs/big-data/announcing-spark-connect-on-amazon-emr-serverless-interactive-pyspark-development-anywhere/

The shape of the thing

Here’s the architecture:

1. A small Python runner calls the EMR Serverless session APIs to start an interactive session and fetch an endpoint hostname plus a short-lived auth token.
2. It exports SPARK_REMOTE, and the stock dbt-spark session method connects over Spark Connect (a sc:// URL, gRPC over TLS on port 443) and runs your models. The remote Spark driver reads and writes S3 Tables using the EMR Serverless execution role. (From a laptop that gRPC goes to the public endpoint; from an MWAA worker it goes through a VPC endpoint — more on that later.)

The dbt project itself is a classic jaffle-shop layout — raw seeds, staging models, and marts — but every table is Iceberg living in S3 Tables.

The star of the show: SPARK_REMOTE and the stock adapter

Spark Connect decouples your client from the Spark driver. Your laptop sends the logical plan; a remote driver executes it. EMR Serverless exposes this as managed interactive sessions: you call an API, it provisions a driver and executors, and hands you back an endpoint and a short-lived (~1 hour) token that you present in a header called x-aws-proxy-auth. The connection URL ends up looking like this:

sc://<host>:443/;use_ssl=true;x-aws-proxy-auth=<token>

PySpark’s SparkSession.builder.getOrCreate() honors an environment variable, SPARK_REMOTE — set it to that sc:// URL and PySpark opens a remote Spark Connect session. And dbt-spark's plain session method is nothing more than getOrCreate(). So the released dbt-spark[session] adapter connects to EMR Serverless. This is documented in Using dbt-core with EMR Serverless, the profile used in dbt should like this:

# dbt/profiles.yml
emr_serverless:
  target: dev
  outputs:
    dev:
      type: spark
      method: session
      host: NA          # required by the schema, ignored when SPARK_REMOTE is set
      schema: "{{ env_var('DBT_SPARK_SCHEMA', 'analytics') }}"

Who sets SPARK_REMOTE? A small runner that owns the session lifecycle:

# runner/run_dbt.py (abridged)
session = EmrServerlessSparkConnect(application_id, region)
session.ensure_application_started()
session.start_session(execution_role_arn, runtime_configuration=spark_conf)
session.wait_until_ready()
endpoint = session.get_endpoint()
env["SPARK_REMOTE"]     = endpoint.spark_remote_url   # sc://...;x-aws-proxy-auth=<token>
env["DBT_SPARK_SCHEMA"] = schema
subprocess.run([dbt, *dbt_command, "--project-dir", project_dir], env=env)
# ... finally: session.terminate()

Start a session, export the endpoint, run dbt, and terminate so the meter stops. Because the token lasts about an hour, each invocation gets a fresh session — perfectly fine for interactive development and scheduled batch runs alike.

Two footnotes worth internalizing. First, the client’s PySpark version must match the remote Spark version (4.0.2 on emr-spark-8.0.0), so we pin pyspark[connect]==4.0.2 — a version mismatch is the classic Spark Connect footgun. Second, the session method calls enableHiveSupport(), which Spark Connect politely refuses (CANNOT_MODIFY_CONFIG) — a harmless warning, since our S3 Tables catalog is a named default catalog configured on the remote session, not the Hive one.

S3 Tables, and the CTAS gotcha that shaped the whole design

Amazon S3 Tables is a purpose-built S3 bucket type that stores Apache Iceberg tables and handles compaction and snapshot cleanup for you. It’s an ideal dbt target: dbt creates and mutates tables, S3 Tables keeps them healthy.

There are two ways to connect Spark to S3 Tables, and choosing between them was the single most consequential decision in this build.

Option A — the Iceberg REST endpoint. S3 Tables ships a standard Iceberg REST catalog at https://s3tables.<region>.amazonaws.com/iceberg, authenticated with SigV4. It's clean and needs no extra JARs. But it has a limitation buried in the docs: it does not support CREATE TABLE AS SELECT yet.

That single line is a dealbreaker for dbt. dbt’s default table materialization is a CTAS. Point dbt at the REST catalog and your first dbt run fails.

Option B — the S3 Tables Catalog for Apache Iceberg client. This catalog implementation (software.amazon.s3tables.iceberg.S3TablesCatalog) talks to the S3 Tables APIs directly and does support CTAS. So that's what the example uses. The catalog config is injected into every session as Spark runtime configuration:

# rendered into the session's spark-defaults
spark.sql.catalog.s3tablesbucket             = org.apache.iceberg.spark.SparkCatalog
spark.sql.catalog.s3tablesbucket.catalog-impl = software.amazon.s3tables.iceberg.S3TablesCatalog
spark.sql.catalog.s3tablesbucket.warehouse   = arn:aws:s3tables:<region>:<acct>:bucket/<name>
spark.sql.defaultCatalog                     = s3tablesbucket
spark.sql.extensions                         = org.apache.iceberg.spark.extensions.IcebergSparkSessionExtensions

Setting defaultCatalog means dbt's plain schema.table references resolve straight into S3 Tables — dbt doesn't even need to know it's talking to a special catalog.

There’s a second S3 Tables quirk that only bites on the second run. S3 Tables refuses to drop a table without purge (dropTable(purge=false) → error), but dbt's default drop — used by seed full-refresh and table replacement — is a plain DROP TABLE IF EXISTS. So the first dbt build sails through on an empty namespace, and the next one fails trying to replace what's already there. The fix is a one-macro override that makes table drops use PURGE:

-- dbt/macros/drop_relation.sql
{% macro spark__drop_relation(relation) -%}
  {%- if relation.type == 'table' -%}
    {% call statement('drop_relation', auto_begin=False) -%}
      drop table if exists {{ relation }} purge
    {%- endcall %}
  {%- else -%}
    {% call statement('drop_relation', auto_begin=False) -%}
      drop {{ relation.type }} if exists {{ relation }}
    {%- endcall %}
  {%- endif -%}
{%- endmacro %}

Spark 4.0.2 honors the PURGE, and re-runs become idempotent. If you only ever test the happy path once, you'll ship this bug.

The lesson: when you’re wiring a SQL tool to a catalog, check whether CTAS, MERGE, and drops behave the way your tool assumes , and always run your pipeline twice.

Reaching Iceberg’s table features from a dbt config()

Once dbt is talking to S3 Tables, the interesting question becomes: how much of Iceberg can you drive from a model, without writing custom macros? More than you’d expect. dbt-spark forwards config keys straight into the SQL it generates — partition_by becomes PARTITIONED BY (...), tblproperties becomes TBLPROPERTIES (...), and incremental_predicates gets ANDed into the merge ON clause. That's enough to reach hidden partitioning, per-column statistics, and partition-pruned merges. Here's the fct_orders model:

{{
    config(
        materialized="incremental",
        incremental_strategy="merge",
        unique_key="order_id",
        file_format="iceberg",
        partition_by="days(order_date)",
        incremental_predicates=[
            "DBT_INTERNAL_DEST.order_date >= date_add(current_date(), -3)"
        ],
        tblproperties={
            "write.metadata.metrics.default": "counts",
            "write.metadata.metrics.column.amount": "full",
            "write.metadata.metrics.column.status": "full"
        }
    )
}}

partition_by is a partition transform, not a column. Because this is Iceberg, days(order_date) is hidden partitioning: Iceberg derives a partition value from order_date and prunes on it automatically, so a query that filters where order_date between ... gets pruning without anyone adding an order_date_day column or writing WHERE partition_col = .... The full transform vocabulary is available — years(), months(), days(), hours(), bucket(N, col), truncate(N, col) — and because the spec is hidden you can evolve it later without rewriting history. The one judgment call is granularity: days() on twenty rows is overkill (you get one tiny file per day), so reach for months() or years() on sparse data and save days()/hours() for real volume.

tblproperties is where per-column statistics live. The write.metadata.metrics.* keys tell Iceberg how much column-level metadata to record in each manifest — lower/upper bounds and value/null counts — which the planner then uses for scan pruning and predicate pushdown. Setting default = counts keeps every column cheap, while full on amount and status collects bounds for the columns you actually filter and aggregate on. It's the same dict-to-TBLPROPERTIES bridge, so any Iceberg write property (compression codec, target file size, and so on) is reachable the same way.

incremental_predicates is the merge's ON-clause performance lever. By default dbt builds the merge condition from unique_key alone — DBT_INTERNAL_SOURCE.order_id = DBT_INTERNAL_DEST.order_id — which forces Iceberg to scan every partition of the target looking for matching order_ids. On a large table that's the whole cost of the run. dbt-spark ANDs incremental_predicates into that same ON clause, so adding a bound on the partition column changes the generated SQL to:

merge into analytics.fct_orders as DBT_INTERNAL_DEST
    using fct_orders__dbt_tmp as DBT_INTERNAL_SOURCE
    on DBT_INTERNAL_DEST.order_date >= date_add(current_date(), -3)
           and DBT_INTERNAL_SOURCE.order_id = DBT_INTERNAL_DEST.order_id
    when matched then update set *
    when not matched then insert *

Now Iceberg prunes on order_date and only touches the last few day-partitions — exactly the win you want on a partitioned table. There's a sharp edge, though, and it's worth stating plainly: a predicate in ON doesn't just filter, it gates matching. If a source row's key lives in the target but in a partition outside the window, it won't match and gets inserted as a duplicate. The window has to cover every partition where an update can legitimately land. It's safe in this model because the incremental filter only ever feeds in orders newer than the current max order_date, and the unique test on order_id will scream if that assumption is ever wrong. Widen it per run with --vars '{merge_lookback_days: 30}'.

You don’t have to take the DDL on faith, you can run these iceberg queries that let you inspect the tables:

DESCRIBE TABLE EXTENDED s3tablesbucket.analytics.fct_orders;
-- # Partitioning
-- Part 0   days(order_date)

SHOW TBLPROPERTIES s3tablesbucket.analytics.fct_orders;
-- write.metadata.metrics.default          counts
-- write.metadata.metrics.column.amount    full
-- write.metadata.metrics.column.status    full

SELECT * FROM s3tablesbucket.analytics.fct_orders.partitions;
-- one row per order_date_day, with record_count / file_count / size

One thing to remember operationally, and it’s a dbt limitation rather than an Iceberg one: changing partition_by on a table that already exists is a no-op on a normal incremental run. Iceberg itself supports partition evolution in place — ALTER TABLE ... ADD PARTITION FIELD rewrites no data (old files keep their spec, new writes use the new one) — but dbt-spark only emits the partition spec in the CREATE TABLE DDL, which runs on the first build and on --full-refresh. An incremental run just MERGEs into whatever spec the table already has; it never diffs partition_by against the live table. So --full-refresh is dbt's built-in way to apply a new spec — with the irony that it does so by dropping and recreating the table, a full rewrite, which is the opposite of Iceberg's zero-rewrite evolution. If you want the in-place version, run the ALTER TABLE ... ADD/DROP PARTITION FIELD yourself (a dbt run-operation or a one-off statement) and dbt's subsequent MERGEs will write new data under the new spec automatically.

Note that this may change on future dbt versions.

Terraform can’t enable Spark Connect sessions (yet)

To accept Spark Connect clients, an EMR Serverless application needs interactiveConfiguration.sessionEnabled = true. That field exists in CloudFormation and the AWS API — but at the time of writing, neither the aws nor the awscc Terraform provider exposes it. The aws provider's interactive_configuration block only knows about the Livy and Studio toggles.

The pragmatic fix: create the application with Terraform, then flip the flag with the AWS CLI in a null_resource. UpdateApplication is a partial update and works while the freshly created app is stopped.

resource "null_resource" "enable_spark_connect_sessions" {
  triggers = { application_id = aws_emrserverless_application.this.id }
  provisioner "local-exec" {
    command = <<-EOT
      aws emr-serverless update-application \
        --application-id "${aws_emrserverless_application.this.id}" \
        --interactive-configuration '{"sessionEnabled":true}'
    EOT
  }
}

It's a small thing, but it means the AWS CLI has to be present during terraform apply. Worth documenting loudly so nobody's surprised. There's a sharper edge, too: because the provider doesn't own that field, it wipes the interactive configuration whenever it updates the application for any other reason — silently turning Spark Connect off until you notice a Sessions are not enabled error. The fix is to re-assert the flag on every apply (a fast no-op when it's already set), so a stray terraform apply can't quietly break your sessions.

Note that this may change on future terraform aws provider versions.

The dbt project

The models are deliberately familiar so the focus stays on the platform:

• Seeds (raw_customers, raw_orders, raw_payments) become the raw Iceberg tables on S3 Tables — loaded through the EMR Serverless session itself.
• Staging models are ephemeral: they compile to CTEs and create no database objects, which sidesteps questions about Iceberg view support and keeps the namespace tidy.
• Marts are real Iceberg tables. dim_customers is a straightforward table; fct_orders is incremental and shows off Iceberg's row-level MERGE:

{{ config(
    materialized="incremental",
    incremental_strategy="merge",
    unique_key="order_id",
    file_format="iceberg",
) }}

select o.order_id, o.customer_id, o.order_date, o.status,
       coalesce(op.amount, 0) as amount
from orders o
left join order_payments op on o.order_id = op.order_id
{% if is_incremental() %}
  where o.order_date > (select coalesce(max(order_date), date '1900-01-01') from {{ this }})
{% endif %}

Because the whole thing is Spark SQL under the hood, you get Spark 4’s niceties — ANSI mode, the VARIANT type and Iceberg v3 .

Sharing the lineage: a password-protected docs site

A transformation project is only as useful as it is legible to the rest of the team. dbt already solves this: dbt docs generate produces a browsable site with a model lineage graph and a full data catalog — columns, types, descriptions, and stats pulled live from the warehouse. The question is how to share it without standing up a web server or leaving it open to the world.

Because dbt docs generate queries the catalog, it runs over the same Spark Connect session as everything else. The --static flag bundles the whole thing into one self-contained HTML file, which is perfect for static hosting. The runner's --publish-docs flag does the generate-and-upload in one step:

uv run python run_dbt.py --publish-docs --minimal -- docs generate --static
# generates dbt/target/static_index.html (lineage + catalog inlined) and
# uploads it to s3://<docs_bucket>/index.html

The --minimal flag is worth a word: dbt docs generate only introspects the catalog, so the session doesn't need the compute a dbt build does. It shrinks the session's Spark workers to small 2 vCPU driver and executors (half the 4 vCPU default), which keeps the footprint well under the application's maximumCapacity — so a docs refresh is far less likely to trip ServiceQuotaExceededException when it overlaps a running build session.

To publish it, the example uses a tidy little serverless stack: a private S3 bucket holds the HTML, and CloudFront serves it through an Origin Access Control so the bucket itself is never public. The password gate is a CloudFront Function on the viewer-request event — a few lines of JavaScript that run at every edge location and reject anything without the right HTTP Basic Auth header:

function handler(event) {
    var request = event.request;
    var expected = "Basic <base64(user:password)>"; // injected by Terraform
    if (!request.headers.authorization || request.headers.authorization.value !== expected) {
        return {
            statusCode: 401,
            statusDescription: "Unauthorized",
            headers: { "www-authenticate": { value: "Basic realm=\"dbt docs\"" } }
        };
    }
    return request;
}

I reached for CloudFront Functions rather than Lambda@Edge on purpose: they run at all edges with no us-east-1 deployment dance, they're billed in fractions of a cent per million invocations, and Basic Auth is about as much logic as they need. Terraform precomputes the base64("user:password") string, embeds it in the function, and can auto-generate a strong password that it surfaces as a sensitive output. Prefer to gate by IP instead, Swap the header check for a request.clientIp test or bolt an AWS WAF IP set onto the distribution, same shape, different predicate.

One honest caveat: since the credential is baked into the (readable) function code, this is “keep casual visitors out”, not a vault. On produciton you may want to keep this on private network.

Keeping it fresh. Docs are only useful if they track the project, so a second MWAA DAG — dbt_docs_refresh — regenerates and republishes the site on a daily schedule (and on demand, from the Airflow UI). It's the same run_dbt.py --publish-docs under the hood, so there's one code path for laptop and scheduler.

One bootstrapping wrinkle is worth calling out: on a brand-new stack the tables don't exist yet, so there's no catalog to render — but the CloudFront/S3 wiring still needs something to serve, and terraform apply would otherwise fail on the missing file. So Terraform seeds a tiny placeholder page on the first apply and marks the docs object with ignore_changes, so later applies never clobber a freshly generated site. Placeholder first, real docs on the next refresh.

You can get the generated password inspecting ouputs from terraform.

terraform output docs_url
terraform output -raw docs_password

Takeaways

A few things I’ll carry into the next project:

• Spark Connect turns a Spark cluster into an endpoint. Combined with EMR Serverless, “run dbt against the lake” stops being an infrastructure project and becomes a config file.
• Managed Iceberg is a genuine unlock for dbt on a lake. S3 Tables removes the compaction/cleanup toil that usually makes lakehouse dbt a part-time job.
• Check catalog capabilities before you commit. The REST-vs-client-catalog CTAS difference , and the drop-needs-PURGE quirk,would have been nasty surprises mid-build. Run your pipeline twice.
• Sharing results can be serverless too. A private S3 bucket, CloudFront, and a few lines of edge JavaScript turn dbt docs into a password-protected lineage site for pennies ,no web server, no open bucket.

The full, validated example: Terraform, the dbt project, the runner, and the Airflow DAG is in the repository. Clone it, point it at a sandbox account, and you’ll have dbt transforming Iceberg tables on serverless Spark in about the time it takes MWAA to spin up.

Continuous Processing took a swing at this back in 2.3 but stayed experimental and supported only a thin set of map-like operations.

Real-time Mode (RTM), introduced in Apache Spark 4.1.0, is the proper second attempt — and this time it keeps the same DataFrame/SQL API and the same processing guarantees as the engine you already know.

This post is a hands-on tour with a companion github repo that you can run locally: a Spark 4.1 cluster, a single-node Kafka 4.1 broker, and a notebook that reads a 10,000 events/sec stream, applies a stateless transformation, and writes it back to Kafka in Real-time Mode.

The code for this post lives in:

GitHub - Neuw84/rtm-spark: Examples on Spark RTM Structured Streaming

What Real-time Mode actually is

Real-time Mode is a new streaming trigger. Instead of waking up on an interval to process the next micro-batch, RTM launches long-running tasks — one per input partition — that stay up and pull records continuously.

A record flows source -> transform -> sink as soon as it arrives, rather than waiting for a batch boundary. Databricks, who contributed the feature, describe the goal as processing data in milliseconds without rewriting your pipeline (architecture deep dive).

The headline numbers from the Spark 4.1.0 release notes: continuous, sub-second processing, and for stateless queries latency can drop to single-digit milliseconds. That is one to two orders of magnitude below the micro-batch floor.

What you keep:

• The same API. It is still readStream / writeStream and the same DataFrame operators. RTM is a trigger, not a new programming model.
• The same semantics. Checkpointing and the write-ahead log still give you fault tolerance and the exactly-once / at-least-once guarantees of the sink.

What changes is how the query is executed underneath that API.

A word on versions

The release notes say Real-time Mode shipped in 4.1.0, and that is true — but getting the PySpark path to actually run took some version archaeology. Here is the lay of the land as I found it:

• 4.1.0 — RTM is announced and present on the JVM, but I could not get the trigger to run end-to-end cleanly.
• 4.1.1 — works from Scala, but the Python binding for the trigger is not wired up yet.
• PySpark, natively — you need to move to a preview build: pyspark==4.2.0.dev5 is the first one where DataStreamWriter.trigger(realTime=...) exists in Python. If you want first-class Python support, that is the version to install.
• Stateful RTM — aggregations, stream-stream joins, dedup, anything that keeps state — is targeted for Spark 4.3. Today RTM is stateless-only.

This demo pins the cluster to Spark 4.1.2 and reaches the trigger from Python through a tiny py4j bridge (more on that below), so it runs on the stable release without the preview wheel.

If you would rather call it natively from Python, install the 4.2.0.dev5 preview and use trigger(realTime="10 seconds") directly instead of the bridge.

How the demo is wired

The companion stack is intentionally small but complete:

• Spark 4.1.2 standalone cluster — a master and two workers (4 cores each, so 8 worker cores total), plus a history server, all built on Ubuntu 24.04 / Python 3.12.
• Apache Kafka 4.1.0, single node, KRaft mode (no ZooKeeper). Broker and controller in one process, PLAINTEXT on kafka:9092 for the Docker network and a PLAINTEXT_HOST listener on localhost:29092 so you can reach it from your laptop.
• JupyterLab as the Spark driver. It runs natively on your host architecture (the cluster talks JVM-to-JVM, so the driver’s CPU arch is irrelevant), which avoids a nasty ipykernel deadlock under amd64 emulation on Apple Silicon.
• A kafka-init one-shot that creates the events-input and events-output topics with 3 partitions each.

The Kafka Structured Streaming connector jars (spark-sql-kafka-0-10, the token provider, kafka-clients, commons-pool2) are baked straight into the Spark and Jupyter images, so there is no spark.jars.packages .

One sizing rule matters for RTM: because it runs one long-lived task per input partition, you need at least one core per partition, held for the life of the query. The input topic has 3 partitions and the cluster has 8 worker cores, so we have headroom. If you bump partitions past your core count, some readers will never get scheduled and the query stalls.

Quick start

git clone https://github.com/Neuw84/rtm-spark.git
cd rtm-spark
docker compose up --build

Then open JupyterLab at http://localhost:8888 (the token is disabled for local convenience — secure it before using this anywhere shared) and open realtime_streaming.ipynb. The notebook and the producer script are bind-mounted into the Jupyter working directory, so they are already there.

The pipeline, cell by cell

1. Produce ~10,000 events/sec

The notebook starts a background producer thread (the same logic is in kafka_producer.py if you would rather run it in a terminal). Each event is a small JSON record keyed by user_id so a user's events land on one partition:

evt = {
    "event_id": seq,
    "user_id": random.randint(1, 100_000),
    "event_type": random.choice(["click", "view", "purchase", "add_to_cart", "logout", "login"]),
    "country": random.choice(["US", "ES", "DE", "FR", "BR", "IN", "JP", "GB"]),
    "amount": round(random.uniform(0.0, 500.0), 2),
    "ts": time.time(),
}

2. Read the stream and apply a stateless transform

Every operator here is map-like — from_json, filter, withColumn — which is exactly what makes the query eligible for Real-time Mode:

parsed = (
    raw.selectExpr("CAST(value AS STRING) AS json_str")
       .select(from_json(col("json_str"), event_schema).alias("e"))
       .select("e.*")
)

transformed = (
    parsed
      .filter((col("event_type").isin("purchase", "add_to_cart")) & (col("amount") > 0))
      .withColumn("amount_with_tax", (col("amount") * lit(1.21)))
      .withColumn("country", upper(col("country")))
      .withColumn("processed_at", current_timestamp())
)

We keep only revenue-bearing events, add a tax-inclusive amount, normalize the country code, and stamp a processing time. No aggregations, no joins, no state and one stage (clean fit for RTM).

3. Enable Real-time Mode (the py4j bridge)

This is the one piece that is version-sensitive. On the stable 4.1.2 release the PySpark trigger() wrapper exposes processingTime, once, continuous, and availableNow — but not realTime. The JVM has it, so we reach through py4j and attach Trigger.RealTime(...) to the underlying Java DataStreamWriter:

writer = (
    out.writeStream
      .format("kafka")
      .option("kafka.bootstrap.servers", KAFKA_BOOTSTRAP)
      .option("topic", OUTPUT_TOPIC)
      .option("checkpointLocation", CHECKPOINT_DIR)
      .outputMode("update")            # required by Real-time Mode
)

Trigger.RealTime("<duration>") — the duration is a CHECKPOINT interval (min 5s), NOT a latency target. Records still flow within milliseconds.

_jvm = spark._sc._jvm
_real_time_trigger = _jvm.org.apache.spark.sql.streaming.Trigger.RealTime("10 seconds")
writer._jwrite.trigger(_real_time_trigger)

query = writer.start()

If you are on the 4.2.0.dev5 preview, you do not need the bridge. Python gets the trigger natively, so you can write it the idiomatic way:

out.writeStream \ 
  .format("kafka") \ 
  .option("kafka.bootstrap.servers", KAFKA_BOOTSTRAP) \ 
  .option("topic", OUTPUT_TOPIC) \ 
  .option("checkpointLocation", CHECKPOINT_DIR) \ 
  .outputMode("update") \ 
  .trigger(realTime="10 seconds") \ 
  .start()

The most common misread: that "10 seconds" is not a latency target. It is the checkpoint cadence — how often RTM commits offsets and progress (minimum 5 seconds). Records keep flowing through the long-running tasks within milliseconds regardless. You are not waiting 10 seconds for output.

The constraints

• Stateless / map-like only. No aggregations, windows, joins, or dedup yet, those are the Spark 4.3 story.
• Output mode must be update.
• A checkpointLocation is required, same as any production streaming query.
• Adaptive Query Execution (AQE) is not supported .
• One core per input partition, held continuously. Size your cluster accordingly.
• The trigger duration is a checkpoint interval, not a latency knob.

Reading the latency numbers

Once the pipeline is running it is tempting to ask “so how fast is it really?” If you attach a probe that timestamps records on the way in and on the way out, you will get a distribution rather than a single number. A run on this local stack looked like:

pipeline (produce -> output write)   p50=25   p95=1299  p99=2026  max=2365  (ms)
end-to-end (produce -> consume)      p50=31   p95=1321  p99=2040  max=2376  (ms)

How to read it:

• p50 (median) — half of all records were faster than this. ~25–31 ms here.
• p95 / p99 (the tail) — 95% / 99% of records were faster than this value. The tail is what users actually feel, and it is where streaming systems live or die.
• max — the single worst record in the window.
• pipeline measures produce -> written to the output topic. end-to-end adds the consumer read on the other side, so it is always a touch higher.

The median in the tens of milliseconds is the RTM story working — comfortably under the micro-batch floor. The fat tail (p99 ~2 s) is not representative of RTM on real hardware; it is the demo environment: everything (broker, master, two workers, driver) crammed onto one laptop, JSON serialization on the hot path, a Python producer, and on Apple Silicon some amd64 emulation for the workers.

Where this leaves us

Real-time Mode is the most interesting thing to happen to Spark streaming in years. It finally attacks the micro-batch latency floor head-on, and it does so without asking you to learn a new API or give up Spark’s fault-tolerance guarantees.

The catch today is scope: stateless queries only, update mode only, and a Python story that is still settling — native PySpark support means the 4.2.0.dev5 preview build, while the stable 4.1.2 release needs the small py4j bridge shown above. Stateful Real-time Mode is the 4.3 milestone to watch.

If your pipeline is a stateless transform-and-route — enrich, filter, reshape, fan out to Kafka or another sink — you can try this today with the companion folder and watch records move in milliseconds instead of batches.

As always: benchmark with your own data, your own hardware, and your own latency requirements before drawing conclusions.

Meanwhile, a generation of native, vectorized SQL engines written in C++ and Rust (ClickHouse, Apache DataFusion, Databricks’ Photon, Velox, and others) execute columnar plans that comfortably outrun Spark’s JVM engine on compute-heavy workloads.

In a previous post I covered Gluten with a Velox backend ( now updated to latest versions). This time we look at a different contender from the same “accelerate Spark without rewriting your code” family: Apache DataFusion Comet.

The code for this post is available on GitHub:

GitHub - Neuw84/spark-comet-iceberg: Demo on Spark acceleration with Comet

Why Comet?

Comet is a Spark plugin that offloads compute-intensive Spark SQL operators to a native engine built on Apache DataFusion, while keeping Spark in charge of distributed scheduling, shuffle coordination, and fault tolerance. Its design goals will feel familiar if you have used Gluten:

- Drop-in plugin: enable it through configuration (`spark.plugins`) — your DataFrame/SQL code stays the same.

- Native operators: scans, projections, filters, aggregations, joins, and sorts are executed in Rust over Apache Arrow columnar batches.

- Native shuffle: Comet ships its own accelerated shuffle (a native implementation and a JVM-based columnar fallback) so shuffle boundaries stay in columnar format.

- Graceful fallback: any operator Comet does not support is transparently handed back to vanilla Spark, so queries always run.

Where Gluten leans on Velox (C++), Comet leans on DataFusion (Rust). Both translate Spark’s physical plan into a native plan and execute it off-heap, but they are independent projects with different operator coverage, memory models, and tuning knobs.

What is Apache DataFusion?

DataFusion is an extensible, high-performance query engine written in Rust and built on Apache Arrow. It provides the reusable building blocks of a modern analytical engine, in the spirit of the composable data stack:

- Arrow-native vectors: a columnar memory layout with zero-copy interchange across the ecosystem.

- Vectorized execution: a fully vectorized evaluator for scalar, aggregate, and window functions.

- Relational operators: scan, filter, projection, hash aggregate, hash/sort-merge join, sort, limit, and more.

- I/O and formats: first-class Parquet support plus pluggable object-store access (S3, GCS, Azure, local).

- Resource management: memory pools, spilling, and multi-threaded execution via a Tokio runtime.

Comet embeds DataFusion as its execution core, so Spark gets DataFusion’s native performance while retaining Spark’s distributed control plane.

Demo project: three storage scenarios

The companion repository is a self-contained, S3-backed environment (Docker Compose plus JupyterLab) very similar in shape to the Gluten/Velox demo: a Spark 4.1.0 cluster with Hadoop 3.4.1, a Spark master, two workers, a history server, and a JupyterLab UI. The difference is the accelerator (Comet 0.16.0) and, more importantly, that it benchmarks three distinct read/write scenarios so you can see where acceleration helps and where the table format dominates:

1. Parquet -> Parquet: the classic datalake files path. Read Parquet from S3, run the analytics queries, and write results back to Parquet. This isolates raw compute and Parquet I/O, and is the most direct comparison to the Gluten/Velox post.

2. Parquet -> Iceberg: an ingestion / transformation path. Read Parquet sources and write the query results into Apache Iceberg tables registered in the AWS Glue Data Catalog. This shows what happens when the write side goes through Iceberg’s writers and catalog commits. And usually found on bronze layer ( medallion architecture).

3. Iceberg -> Iceberg: a fully table-format-native path. Read from Iceberg tables and write results back to Iceberg. This reflects a mature lakehouse where everything is already managed tables with snapshots, schema evolution, and catalog metadata or silver gold layers.

Each scenario runs the same three representative SQL workloads twice, once with Comet enabled and once with vanilla Spark , so the wall-clock comparison is apples-to-apples within a scenario.

Benchmarks

All three scenarios were run on a single AWS EC2 c5.4xlarge instance (16 vCPUs, 32 GB RAM)with the whole cluster: Spark master, two workers, and JupyterLab co-located in Docker on that one box. Because everything runs on a single host, shuffle never crosses the network; on a real multi-node cluster the network cost of shuffling data around would shift these numbers, so treat them as a relative compute comparison rather than an absolute production benchmark.

Each scenario runs the same three queries with Comet ON and then OFF, and we report the wall-clock time per query plus the speedup (Comet OFF / Comet ON, so higher is better). The three queries deliberately span a range of complexity:

• Q_A is a light single-table multi-aggregation dominated by scan and I/O.
• Q_B is a medium rollup over a join
• Q_C is a star-schema join with a distinct count and a top-K sort which is exactly the kind of compute-bound work native engines accelerate best.

Note that for this tests we didn´t enable native Parquet writer as it is yet experimental. Moreover work is also progressing towards having native Iceberg writer.

Seems that the native Iceberg readers are not working as expected. I opened an Issue.

Summary

Apache DataFusion Comet brings Rust-based, Arrow-native vectorized execution to Apache Spark without changing your DataFrame or SQL code. By translating Spark’s physical plan and offloading to DataFusion off-heap, it pairs Spark’s distributed scheduling with a modern native engine — the same value proposition as Gluten/Velox, but a different engine with its own strengths and tuning with a more direct Iceberg integration provided by iceberg-rust.

The companion repository lets you reproduce all of this locally.

As always: benchmark with your own data and your own queries before drawing conclusions.

But not every analytics problem is “big data.” In fact, most workloads don’t need a large distributed cluster.

That’s where DuckDB shines. Often called the “SQLite for analytics,” DuckDB is an embedded OLAP database that runs blazingly fast on a single machine. With vectorized execution, efficient caching, and zero server overhead, DuckDB handles surprisingly large datasets interactively. Think: data exploration, prototyping, dashboards, or any workload where the startup overhead of Spark would be overkill.

And instead of wiring all of this up manually using Amazon S3 bucket with Glue as catalog, we are going to use Amazon S3 Tables. S3 Tables deliver the first cloud object store with built-in Apache Iceberg support and streamline storing tabular data at scale with automatic maintenance. This means that tools like DuckDB can plug into Iceberg on S3 Tables with just an ARN, and instantly interoperate with other engines (Athena, EMR, Spark, Flink, etc.).

In this post, I’ll walk you through an end-to-end example of writing data into S3 Tables using DuckDB 1.4.0 with nightly extensions. All you need is:

• An S3 Tables bucket in your AWS account.
• A namespace called duckdb ont that bucket(created via the AWS console).
• The ARN of your S3 Tables bucket.

Let’s get started.

Open the DuckDB CLI by running:

duckdb

Step 1. Install the Extensions

DuckDB ships with a rich extension system. Since Iceberg support is still stabilising, we’ll use nightly builds of the extensions:

FORCE INSTALL aws FROM core_nightly;
FORCE INSTALL httpfs FROM core_nightly;
FORCE INSTALL iceberg FROM core_nightly;

Step 2. Load AWS and Set Up Credentials

We’ll use the built-in AWS extension to authenticate via the standard credential chain (env vars, IAM roles, or config files):

LOAD aws;
CREATE SECRET (
      TYPE s3,
      PROVIDER credential_chain
);

Step 3. Attach the Catalog

Next, attach your S3 Tables backed Iceberg catalog. Replace <<CUSTOM_ARN>> with the ARN of your S3 table bucket:

ATTACH '<<CUSTOM_ARN>>' AS s3_tables (
   TYPE iceberg,
   ENDPOINT_TYPE s3_tables
);

Step 4. Validate the Connection

Check that your catalog is live:

SHOW ALL TABLES;

If your namespace duckdb is configured correctly, you’ll see it listed.

Step 5. Create Synthetic Data

Let’s create a table by pulling in some real CSV data from the DuckDB public datasets:

CREATE TABLE s3_tables.duckdb.services AS
      FROM 'https://blobs.duckdb.org/nl-railway/services-2025-03.csv.gz';

This writes an Iceberg table into our S3 Table bucket directly from DuckDB.

Step 6. Explore with the DuckDB UI

DuckDB also ships with a simple web UI for interactive exploration:

CALL start_ui();

Open the provided URL in your browser and start querying your freshly created Iceberg table.

You can create a notebook a issue this query to see the inserted data. Note that you need to change the database to memory on each cells as shown in below image ( use this for all the cells you run).

from s3_tables3.duckdb.services
select
 "Service:RDT-ID",
 "Service:Date",
 "Service:Type",
 "Service:Company",
 "Service:Train number",
 "Service:Completely cancelled",
 "Service:Partly cancelled",
 "Service:Maximum delay",
 "Stop:RDT-ID",
 "Stop:Station code",
 "Stop:Station name",
 "Stop:Arrival time",
 "Stop:Arrival delay",
 "Stop:Arrival cancelled",
 "Stop:Departure time",
 "Stop:Departure delay",
 "Stop:Departure cancelled"
limit 100

This will show the data that we have previously created. But we are here to write some data too. Let’s create a table from our existing data.

CREATE TABLE s3_tables3.duckdb.services_per_month AS
     SELECT
         month("Service:Date") AS month,
         "Stop:Station name" AS station,
         count(*) AS num_services
     FROM s3_tables3.duckdb.services
     GROUP BY ALL;

Then we can visualize our newly created data, that would be available also for any compliant Iceberg engine. For example: Amazon Athena, AWS Glue, Amazon EMR, Spark, Flink,…etc.

SELECT * FROM s3_tables3.duckdb.services_per_month;

You can explore the column statistics on the Duckdb ui along with the capability to download query results.

Why This Matters

This workflow demonstrates the sweet spot of DuckDB: quick, local ( or remote if you wish) analytics with the ability to interoperate with your LakeHouse through Iceberg. You don’t need to spin up Spark or EMR to publish clean, structured datasets back to S3 or S3 Tables.

• Speed: DuckDB is optimized for vectorized columnar execution.
• Simplicity: No servers, no daemons, just a single binary.
• Interoperability: Iceberg tables you write with DuckDB can be read and extended by Spark, Flink, Athena, and beyond.

This makes DuckDB an increasingly important part of the modern analytics ecosystem: not a replacement for Spark, but a perfect companion for “small data” and interactive workloads.

Meanwhile, several native SQL engines (ClickHouse, Apache DataFusion, Databricks’ Photon, Sail and Velox) implement columnar, vectorized execution in C++ and can outperform Spark’s JVM‑based engine. To make Spark scale with the speed of native engines, the open‑source Gluten project “glues” them together.

Why Gluten?

Gluten is a middle layer that offloads compute‑intensive Spark SQL operators to native engines while preserving Spark’s distributed control flow. Its design focuses on the following:

• Plan conversion: Spark’s whole‑stage physical plan is translated to a Substrait plan and passed to the native engine.
• Offload execution: Performance‑critical data processing is executed in native code. Gluten exposes clear JNI interfaces so Spark can call the native backend and get data back as Arrow columnar batches.
• Memory & shuffle: Gluten manages its own off‑heap memory and implements a columnar shuffle manager; it falls back to vanilla Spark for unsupported operators.
• Backend flexibility: The plugin can switch native backends (currently Velox and ClickHouse) without changing Spark applications.

The goal is to let users keep their DataFrame/SQL API unchanged and simply enable the plugin via configuration to accelerate Spark’s core computation.

What is Velox?

Velox is an open‑source C++ execution engine that provides reusable, high‑performance building blocks for data systems. Developed by Meta and maintained with partners like IBM/Ahana, Intel, Voltron Data, Microsoft and ByteDance, Velox takes an optimized query plan (from a SQL engine or dataframe API) and executes it. It does not include its own SQL parser, dataframe API or optimizer; instead it offers a library of components that compute engines can integrate ( remember, Composable Data Stack):

• Type and vector system: An Arrow‑compatible columnar memory layout with encodings like flat, dictionary and run‑length.
• Vectorized expression evaluation: A fully vectorized evaluator for scalar, aggregate and window functions.
• Operators & functions: Built‑in relational operators (scan, projection, filter, join, group by, order by, unnest, etc.).
• I/O and connectors: Interfaces for multiple file formats (Parquet, ORC, Nimble) and storage systems (S3, HDFS, GCS, ABFS, local files).
• Resource management: Memory arenas, buffer management, CPU thread pools, spilling and caching

By using Velox as a backend, Gluten can offload Spark’s physical operators to a highly optimized native engine while maintaining Spark’s fault tolerance and distributed scheduling.

Demo project: spark‑gluten‑velox

To illustrate how Gluten with a Velox backend accelerates Spark workloads, the repository Neuw84/spark-gluten-velox provides a Docker‑Compose deployment and a test notebook. The repo is a self‑contained environment built for S3‑backed benchmarks:

• Components: It assembles a Spark 3.5.2 cluster with Hadoop 3.4.1, Gluten 1.4.0 using the Velox backend, JupyterLab for interactive notebooks, and S3A integration. The images are based on Ubuntu 22.04.
• Cluster layout: A Spark master (port 8080), two worker nodes (4 cores & 4 GB RAM on-heap each), a Spark history server (port 18080) and a JupyterLab interface (port 8888)GitHub.
• Requirements: You need Docker and Docker Compose on x86‑64 hardware; an AWS S3 bucket for reading/writing data; and either an IAM role (recommended on EC2) or AWS credentials on your workstation configured.

Quick start

• Clone and build the environment: docker Compose builds images for the master, workers, history server and JupyterLab, and starts the cluster.

git clone https://github.com/Neuw84/spark-gluten-velox.git
cd spark-gluten-velox
docker-compose up - build

• Open JupyterLab. Navigate to http://localhost:8888 in your browser. Authentication is disabled for convenience; upload the test.ipynb notebook provided in the repo.
• Configure S3 access. On EC2 you can attach an IAM role that grants s3:GetObject, s3:PutObject, s3:DeleteObject and s3:ListBucket permissions on your bucket. For local development configure your AWS credentials via aws configure or environment variables (AWS_ACCESS_KEY_ID, AWS_SECRET_ACCESS_KEY) and update the s3_base value in the notebook.
• Run the benchmark. Upload test.ipynb to the JupyterLab and execute its cells. It will first run queries with Velox enabled and then with vanilla Spark for comparison. Remember to configure the S3 bucket.

What the notebook does

The test notebook synthesizes a 50 million‑row fact table and small dimension tables to simulate a retail analytics workload. It defines constants such as number of products (N_PROD), countries, channels and days, and uses PySpark’s random functions to generate skewed data. After caching the fact and dimension tables, it defines a simple timing function that writes each query result to S3 and counts the rows back to ensure full execution. The notebook runs three representative SQL workloads

• Query A — Heavy multi‑aggregation: Groups the fact table by country_id and channel_id and computes counts, sums, averages, standard deviation and approximate percentiles. This pattern stresses hash aggregation, projection and filter operators.
• Query B — Rollup (cube) aggregation: Joins the fact table with a date dimension and uses rollup(date_key, country_id, product_id) to compute revenue, quantity and average discount across multiple grouping levels.
• Query C — Star schema join and top‑K sort: Joins the fact table with broadcast dimensions (countries and channels) and the date dimension, computes gross and net revenue, and orders by gross descending, taking the top 5,000 rows. Broadcast joins and top‑K sorts test Velox’s vectorized join and sort operators.

After executing the queries, the notebook prints wall‑clock times and resets spark.gluten.enabled to false to run the same workload with the vanilla Spark engineGitHub.

Configuration highlights

To enable Gluten/Velox acceleration, the notebook constructs a SparkSession with several critical settings:

• Enable the Gluten plugin: spark.plugins = org.apache.gluten.GlutenPlugin.
• Use the columnar shuffle manager: spark.shuffle.manager = org.apache.spark.shuffle.sort.ColumnarShuffleManager.
• Turn on off‑heap memory and allocate 8 GB: spark.memory.offHeap.enabled = true and spark.memory.offHeap.size = 8g.
• Configure the Velox bundle JAR on the driver and executors: the Docker image already includes gluten-velox-bundle-spark3.5_2.12-linux_amd64-1.4.0.jar and adds it to the classpath in the notebook.

For S3 integration, the compose file and notebook use the Hadoop 3.4.1 S3A connector with “magic” committers, improving write performance. You can view progress and metrics in the Spark UIs (master at :8080, history server at :18080, and driver UI at :4040).

Customizing and scaling

The demo is designed to be flexible:

• Scaling workers: Adjust SPARK_WORKER_CORES and SPARK_WORKER_MEMORY in the docker-compose.yml to give executors more cores and memory.
• Adding workers: Duplicate the worker service and assign different ports.
• Changing data size: Modify N_ROWS, REPART or other parameters in the notebook to simulate different workload sizes.
• Custom queries: Feel free to add your own workloads or business logic to the notebook; the notebook’s timing function is generic.
• Enhancing the compose setup: The current compose file keeps event logs local; you can improve it to store logs on S3 so the Spark history server works across runs and workers.

Benchmarks

Let’s do a concrete benchmark. We will run each set 3 times and do the average with different test sizes. We will run them on a c5.4xlarge instance. Remember as we are running all containers on a single instance we are not touching real network. On a real production scenario this could impact the results ( shuffling data around the network is not cheap). Test parameters:

Default settings:

N_ROWS = 50_000_000 
N_PROD = 50_000
N_CTRY = 32
N_CH = 6
N_DAYS = 90
REPART = 256

Medium settings:

N_ROWS = 80_000_000   
N_PROD = 70_000
N_CTRY = 32
N_CH   = 6
N_DAYS = 120
REPART = 256         

High settings:

N_ROWS = 110_000_000
N_PROD = 80_000
N_CTRY = 32
N_CH   = 6
N_DAYS = 120
REPART = 256

Then, let’s see the results for each query:

Query A:

Query B

Query C

Velox reduces total wall‑time by roughly 23 % compared to Vanilla Spark. Expressed as an average of the relative improvements for each query‑size combination, the overall performance gain is about 25 % faster for Velox versus Vanilla Spark on this specific test.

Summary

Gluten with a Velox backend brings modern vectorized execution to Apache Spark without requiring users to change their DataFrame/SQL code. By translating Spark’s physical plan to a Substrait plan and offloading to Velox, it combines Spark’s robust distributed scheduling with Velox’s native performance and memory efficiency. The spark-gluten-velox repository demonstrates how to set up a local Spark cluster with Gluten/Velox, configure S3 integration, and benchmark performance across realistic analytics workloads. Once you run the notebook and measure the wall‑clock times, you’ll have concrete numbers to quantify how much faster your queries can run on Spark when backed by Velox.

Remember to test for your specific workload!

Now it is time to move this into a productive setup on top of native AWS stack. Remember that you can go the previous blogpost or the Github repo for a more detailed view of the components. For that we will use the AWS services:

• AWS Glue: here we will put our data-producer Python script. This is in charge of inserting data to the database.
• Amazon Aurora PostgreSQL: we are going to Aurora with Postgre compability to deploy our database.
• Amazon Managed Service for Kafka (MSK): we will use MSK to have a managed Kafka broker.
• MSK Connect: we will use this feature of MSK to register both Debezium Source connector and Iceberg Connect Sink.
• S3 Tables: we will use the new launched Apache Iceberg Rest EndPoint (IRC) for S3 Tables as our catalog and will store Iceberg data on top of S3 Tables.

A 100 hundred feet overview of the architecture can be seen on the following picture:

Requirements:

• An Amazon VPC well configured with minimum of 2 private subnets. You can use this guide.
• Deploy a MSK cluster with version 3.6 with 2 nodes ( 1 for each private subnets) with Unauthenticated access Enabled. Although we are on a private subnet on a production setup you should disable this. You need to enable topic auto creation via Cluster Configurations. The corresponding property is:

auto.create.topics.enable=true

• An Amazon Aurora PostgreSQL 17 database. Any size will work, the instance should be deployed into one of the private subnets. Moreover, Debezium needs logical replication to be enabled. You can follow this guide.
• A Python Shell Glue job that will run inside the VPC and with network access (Security Groups) to the database.
• You will need to create a S3 Table bucket where our CDC data will live with Analytics Services integration enabled.
• You need to deploy the MSK Connectors. We are going to deep dive on this next.

You can download the connectors used for the blogpost on the releases page on the Github repository or you can go to the official project to get the latest ones 😀 . If you want to have detailed instructions on how can you build and use a custom plugin and create a MSK Connect conector you can read blog.

Let´s use first the CloudFormation template to automate this deployment. Go to us-east-1 region and open the CF console, upload the CF template available on the repo and wait for deployment, you can chooose some parameters like the Aurora DB username and password. Then, is a matter of:

1. Create a S3 Table bucket while enabling analytics integration. Take note of the arn of the table.

Go to the S3 bucket created by the CF template ( by default should be debezium-lakehouse-account-id…..) and create two folders athena for storing Athena query results and connectors that will be used for storing MSK Connect custom plugins.

2. Create the data producer. You should go to Glue console and create a Python shell script job. Choose upload script and upload data-producer.py file from the repo.

Then, you need to change the DB parameters as needed on the script. For getting the Aurora endpoint and port you can go to the RDS console ( take note of the writer instance).

Note that we could also use env. parameters or Glue job parameters ( or AWS Secrets Manager secrets), however for a demo more than enough. Remember that as best practice is better not to store any Username and Password on code.

Next step is clicking on job details and add on Job parameters:

--additional-python-modules : psycopg2-binary

Also, you need to select the Connection created by the CF template that will allow our job to reach the Aurora DB. You can use this screenshot as a guide.

In basic properties you will need to put a name (like data-producer) and select the role name. This will have the name of CF stack followed by GlueRole.

Finally, you can click on Save and then Run the job.

3. You will upload the downloaded connectors to S3 in the connector folder you have previously created and use them to create a custom plugin.

4. Create one for Debezium with these properties where you can see that the only change is the endpoint, user and password. Use Kafka Connect version 3.7 and the default config for workers, choose the role with the CF stack name for it ( should be stackName-MSKConnectRole-xxxx):

connector.class=io.debezium.connector.postgresql.PostgresConnector
database.user=REPLACE_WITH_USERNAME
database.dbname=inventory
tasks.max=1
database.server.name=dbserver1
schema.include.list=public
database.port=3306
plugin.name=pgoutput
topic.prefix=ice
database.hostname=REPLACE_WITH_AURORA_ENDPOINT
database.password=REPLACE_WITH_PASSWORD
table.include.list=public.customers
value.converter=org.apache.kafka.connect.json.JsonConverter
key.converter=org.apache.kafka.connect.json.JsonConverter

3. For the Iceberg Sink Connector you can use this as a guide, like on the previous step, use Kafka Connect version 3.7 and let the other defaults, choose the role with the CF stack name for it ( should be stackName-MSKConnectRole-xxxx):

connector.class=org.apache.iceberg.connect.IcebergSinkConnector
s3.region=us-east-1
iceberg.tables.evolve-schema-enabled=true
tasks.max=1
iceberg.catalog.client.region=us-east-1
iceberg.tables.upsert-mode-enabled=false
iceberg.tables.dynamic-enabled=false
errors.deadletterqueue.context.headers.enable=false
iceberg.tables.auto-create-enabled=true
iceberg.tables=my_database.customers_table
iceberg.catalog.rest.sigv4-enabled=true
value.converter=org.apache.kafka.connect.json.JsonConverter
errors.log.enable=true
key.converter=org.apache.kafka.connect.json.JsonConverter
topics=ice.public.customers
iceberg.catalog.io-impl=org.apache.iceberg.aws.s3.S3FileIO
iceberg.control.commit.interval-ms=60000
iceberg.catalog.uri=https://s3tables.us-east-1.amazonaws.com/iceberg
iceberg.catalog.warehouse=arn:aws:YOURS3TABLE_ARN
iceberg.catalog.rest.signing-name=s3tables
iceberg.control.topic=iceberg-connector-control-v1
iceberg.catalog.type=rest
iceberg.catalog.rest.signing-region=us-east-1

4. With the connectors ready is a matter to wait a little bit and go to Athena console and set the S3 folder that we have previously created athena for storing query results.

Now we can go to Athena query editor and see how our data looks like:

Remember that you can deploy the AWS CloudFormation template to automate the deployment, although more things can be automated that is up to the reader 😉.

Happy Iceberg-ing!