Spark + Iceberg architecture — write, read, procedures, Catalyst

Wiring Spark to Iceberg — the runtime jar and the SQL extensions

Iceberg is not built into Spark. Two things have to be present before any of the rest of this article applies: the iceberg-spark-runtime jar matched to your Spark and Scala versions, and the Iceberg SQL extensions registered on the session. The jar name encodes the pairing, which is the first thing to check when a cluster mysteriously cannot find the format — a runtime built for Spark 3.4 on a 3.5 cluster fails in ways that do not name the version mismatch.

spark-sql \
  --packages org.apache.iceberg:iceberg-spark-runtime-3.5_2.12:1.5.2 \
  --conf spark.sql.extensions=org.apache.iceberg.spark.extensions.IcebergSparkSessionExtensions \
  --conf spark.sql.catalog.lake=org.apache.iceberg.spark.SparkCatalog \
  --conf spark.sql.catalog.lake.type=rest \
  --conf spark.sql.catalog.lake.uri=http://catalog-host:8181 \
  --conf spark.sql.catalog.lake.warehouse=s3://my-lake/warehouse

The extensions matter more than they look. They are a parser and analyzer plugin, and they are what add the CALL statement for stored procedures, the partition-evolution DDL (ALTER TABLE ... ADD PARTITION FIELD), the write-order DDL, and branch and tag management to Spark SQL. Without them, plain reads and appends work fine, which is exactly why the omission survives a smoke test — and then the first maintenance job fails with a parse error on CALL that reads like a syntax mistake in your own SQL. Set the extensions unconditionally, in the cluster defaults, not per job.

Iceberg is implemented as a DataSource V2 connector, so the catalog, table, scan-builder and commit-protocol machinery it plugs into is Spark's, not Iceberg's. That contract — capability declaration, pushdown negotiation, the driver-side commit — is the subject of Spark DataSource V2 architecture; here we only care about what Iceberg does inside it.

spark_catalog or a named catalog — the decision that shapes every query

Spark resolves a three-part identifier catalog.namespace.table by looking up the first part in its catalog registry. Iceberg ships two implementations to register there, and choosing between them is the single highest-consequence configuration decision in the integration.

SparkCatalog registers a new, separate catalog name. Everything under it is Iceberg; nothing else can live there. Tables are addressed as lake.sales.orders, and your existing Hive tables keep resolving through spark_catalog untouched. This is the clean choice for a new lakehouse namespace, and it makes cross-format joins explicit because both catalogs appear in the query text.

SparkSessionCatalog is registered over the built-in session catalog. It handles Iceberg tables itself and delegates everything else to the underlying Hive catalog, so a warehouse containing both Iceberg and ordinary Parquet-on-Hive tables keeps working with unqualified names. This is what you want when converting an existing warehouse in place — and it is required for the in-place migrate procedure, because the source table lives in the session catalog.

# a separate Iceberg-only catalog
spark.sql.catalog.lake                 = org.apache.iceberg.spark.SparkCatalog
spark.sql.catalog.lake.type            = hive
spark.sql.catalog.lake.uri             = thrift://metastore-host:9083
spark.sql.catalog.lake.warehouse       = s3://my-lake/warehouse
spark.sql.catalog.lake.io-impl         = org.apache.iceberg.aws.s3.S3FileIO

# OR: take over the session catalog so existing Hive tables still resolve
spark.sql.catalog.spark_catalog        = org.apache.iceberg.spark.SparkSessionCatalog
spark.sql.catalog.spark_catalog.type   = hive

spark.sql.defaultCatalog               = lake

The type property selects the catalog backend — hive, hadoop and rest are the ones you meet most often. For anything else, including AWS Glue and Nessie, set catalog-impl to the fully qualified class instead; that form is always correct, whereas the short type aliases vary by Iceberg version. One warning that costs people data: hadoop derives the current table state from file names in a directory and commits by creating a new metadata file, which is only atomic on a filesystem with atomic rename. On plain object storage it is not safe for concurrent writers. Use a Hive, REST or Glue catalog for anything shared.

Iceberg also caches loaded table objects per catalog. In a long-lived session — a notebook, a Spark Connect server, a Thrift server — that cache is why a table another job just compacted still reports the old snapshot. spark.sql.catalog.lake.cache-enabled=false trades a little planning latency for always reading current metadata.

The architecture — how a Spark job commits to an Iceberg table

Trace a write end to end. The analyzer resolves lake.sales.orders through the registered catalog and gets an Iceberg table object with its schema, partition spec and properties. Spark plans the write and ships a writer factory to the executors. Each task writes one or more complete Parquet files into the table's data location and returns a small commit message to the driver naming the files it produced and their statistics.

Nothing is visible yet. When every task has reported, the driver performs one Iceberg commit: it builds new manifests describing the added files, assembles a new snapshot, writes a new table metadata file, and asks the catalog to swap the table's pointer from the old metadata to the new one. That swap is the atomic act. A Spark job therefore produces exactly one snapshot regardless of how many tasks it ran, and a job that dies mid-write leaves orphaned Parquet files on storage but no change to the table.

The swap is optimistic, and that is where concurrent writers meet. Iceberg attempts the compare-and-swap; if another writer committed first, it re-reads the current metadata, re-validates its own changes against it, and tries again. commit.retry.num-retries (four by default) and the associated wait properties bound that loop. Appends almost always succeed on retry because they conflict with nothing. Overwrites, deletes and merges can fail permanently with a validation error, because the rows they claimed to be rewriting were changed underneath them — which is the correct outcome, not a bug.

The read side is the mirror image and it runs on the driver. Planning opens the current snapshot, reads the manifest list and then the manifests, and uses partition values plus the per-column lower and upper bounds recorded there to discard files that cannot contain matching rows. Only the surviving files are split into tasks. All of that metadata work happens before a single executor is asked for anything, which is why a table with an unhealthy metadata tree shows up as a long, idle gap at the start of a query.

The diagram below lays the same picture out as surfaces rather than as a sequence: user code on top, the extension and catalog layer beneath it, the write and read paths it drives, the procedures and streaming and merge features hanging off them, and the operational concerns — retention, expiry, concurrent writers — underneath everything. The rest of this article walks those boxes one at a time.

Write paths — DataFrameWriterV2, SQL, and distribution mode

The supported DataFrame entry point is writeTo, the DataFrameWriterV2 API. The older df.write.format("iceberg") path still functions, but it predates catalogs and cannot express table creation, partitioning or properties cleanly.

df.writeTo("lake.sales.orders")
  .using("iceberg")
  .partitionedBy(days($"order_ts"), bucket(16, $"customer_id"))
  .tableProperty("format-version", "2")
  .tableProperty("write.target-file-size-bytes", "536870912")
  .create()

df.writeTo("lake.sales.orders").append()             // add rows
df.writeTo("lake.sales.orders").overwritePartitions() // dynamic overwrite
df.writeTo("lake.sales.orders").overwrite($"order_date" === "2026-08-01")

The SQL surface is the same operations by another name: CREATE TABLE ... USING iceberg PARTITIONED BY (days(order_ts)), INSERT INTO, INSERT OVERWRITE. Note that INSERT OVERWRITE obeys spark.sql.sources.partitionOverwriteMode: under the default static it replaces every partition matching the table's spec, under dynamic only the partitions the incoming data touches. The difference between wiping one day and wiping the table is one config key, and the query text looks identical either way.

The property that decides your file layout is write.distribution-mode. With none, every task writes whatever partitions its input happens to contain, so a thousand tasks spread over thirty partitions can emit thirty thousand files in one commit. With hash, Spark shuffles by the partition key first, so each partition is written by a small number of tasks. With range, it range-partitions by the table's sort order, which is what you want for a sorted table. Defaults have changed across Iceberg releases; set it explicitly on every table you care about rather than inheriting whatever the version chose.

The companion property is write.spark.fanout.enabled. A writer normally requires its input clustered so it can close one partition's file before opening the next; with fanout enabled it keeps several files open at once and accepts unclustered input. That buys you a shuffle-free write at the cost of executor memory, and it is the usual setting for streaming appends where shuffling every micro-batch is not affordable.

MERGE INTO — copy-on-write versus merge-on-read

MERGE INTO is the reason most teams adopt Iceberg on Spark, and its cost profile is entirely governed by three table properties that nothing in the SQL mentions.

MERGE INTO lake.sales.customers t
USING staged_updates s
ON t.customer_id = s.customer_id
WHEN MATCHED AND s.op = 'D' THEN DELETE
WHEN MATCHED THEN UPDATE SET *
WHEN NOT MATCHED THEN INSERT *;

Under copy-on-write — the default — Iceberg finds every data file containing a matched row and rewrites that file in full with the changes applied. Updating a hundred rows spread across a hundred 512 MB files rewrites 51 GB. The write is expensive and the read afterwards is as fast as a plain scan, because the files on disk are simply correct.

Under merge-on-read, Iceberg writes new data files for the inserts and updates plus small delete files marking the superseded rows, and leaves the original data files alone. The write is cheap and roughly proportional to the change size. The cost moves to every subsequent reader, which must apply the delete files while scanning. Merge-on-read requires table format version 2 or later.

ALTER TABLE lake.sales.customers SET TBLPROPERTIES (
  'write.merge.mode'  = 'merge-on-read',
  'write.update.mode' = 'merge-on-read',
  'write.delete.mode' = 'copy-on-write'
);

The three modes are independent, which is more useful than it first appears: frequent small merges from a CDC stream want merge-on-read, while a rare bulk DELETE for a retention policy is better off rewriting files once and leaving nothing for readers to reconcile. The rule of thumb is write frequency against read frequency — merge-on-read is a loan against future scans, and compaction is how you repay it. Choosing it without scheduling compaction is the most common way to make an Iceberg table slower than the Hive table it replaced.

Two operational notes. The ON clause is executed as an ordinary Spark join, so everything you know about broadcast thresholds and skew applies directly; a merge against a small staged batch should be broadcasting it, and EXPLAIN will tell you whether it is. And if a single source row matches multiple target rows, the merge fails with a cardinality error rather than silently picking one — deduplicate the source, do not work around the error.

Reads — pruning, time travel, and incremental scans

Pushdown works the way it does for any V2 source: Spark offers filters, Iceberg accepts the ones it can enforce, and Spark keeps a residual filter for the rest. What is Iceberg-specific is how a filter is enforced. Partition predicates eliminate whole manifests. Predicates on ordinary columns are checked against the lower and upper bounds recorded per column per file, so a filter on a well-clustered column prunes almost everything and the same filter on a randomly distributed column prunes nothing — identical SQL, two orders of magnitude apart, with no signal in the results.

The metadata tables are how you see it. They are queryable as ordinary tables with a suffix on the identifier, and they are the primary diagnostic instrument for the entire integration.

SELECT count(*), avg(file_size_in_bytes)/1048576 AS avg_mb
FROM lake.sales.orders.files;

SELECT committed_at, snapshot_id, operation, summary
FROM lake.sales.orders.snapshots ORDER BY committed_at DESC LIMIT 10;

SELECT * FROM lake.sales.orders.manifests;
SELECT * FROM lake.sales.orders.partitions;
SELECT * FROM lake.sales.orders.refs;

The snapshots summary column is worth reading closely — it records added and deleted file counts and record counts per commit, so a streaming job producing four hundred files per micro-batch is visible there long before it becomes a query-latency incident.

Time travel is SQL-level. SELECT * FROM lake.sales.orders VERSION AS OF 3821550127947089612 reads a specific snapshot; TIMESTAMP AS OF '2026-08-01 00:00:00' resolves the snapshot current at that instant. Named branches and tags can be used in the same position, which is what makes an audit-then-publish workflow expressible: write to a branch, validate it, fast-forward the main reference. The DataFrame equivalents are the snapshot-id and as-of-timestamp read options.

Incremental reads are the other half. Given a pair of snapshot ids, Iceberg can return just the rows appended between them — the basis of cheap downstream refresh without a watermark column. The important limitation is that the plain incremental scan covers appends only; a snapshot produced by an overwrite or a delete is not expressible as a set of new rows. For genuine change data you need the changelog view procedure, which materialises inserts, updates and deletes with a change type, and which requires the SQL extensions.

spark.read
  .option("start-snapshot-id", "3821550127947089612")
  .option("end-snapshot-id",   "7160982374128476109")
  .table("lake.sales.orders")

Hidden partitioning and partition evolution from Spark

A Hive-style table forces the partition into the schema: you store an order_date column derived from order_ts, and every query must filter on both or lose pruning. Iceberg records the transform instead, so the table is partitioned by days(order_ts) and a filter on order_ts alone prunes correctly. The derived column disappears, and with it the entire class of bug where someone filters on the timestamp, gets the right answer, and scans the whole table.

The available transforms are identity, years/months/days/hours on a timestamp or date, bucket(N, col) for hashing a high-cardinality key into a fixed number of buckets, and truncate(N, col) for prefixes and numeric ranges. They compose, and bucket on a join key is what makes a storage-partitioned join possible — if both sides of a join are bucketed the same way, Spark can be told to exploit that layout and skip the shuffle entirely.

Because the spec lives in metadata rather than in directory names, it can change without rewriting data:

ALTER TABLE lake.sales.orders ADD PARTITION FIELD hours(order_ts);
ALTER TABLE lake.sales.orders DROP PARTITION FIELD days(order_ts);
ALTER TABLE lake.sales.orders REPLACE PARTITION FIELD days(order_ts) WITH hours(order_ts);

ALTER TABLE lake.sales.orders WRITE ORDERED BY customer_id, order_ts;

What evolution does not do is rewrite history. The old files keep the old spec; new writes use the new one; planning handles both simultaneously. So a table partitioned by day for two years and by hour since Tuesday will prune beautifully on recent data and coarsely on everything older, and the partitions metadata table showing two specs is the explanation for a query whose cost depends on how far back it reaches. If you need the old data under the new layout, that is a compaction job with a partition filter, not a DDL statement.

The write-order DDL in the last line above is the underrated one. It records a sort order on the table so that every writer — and the sort-strategy compaction described next — clusters rows consistently, which is what makes the column-bound pruning above actually prune.

Compaction — rewrite_data_files and the small-file problem

Every commit adds files and nothing ever removes them implicitly. Streaming appends, frequent micro-batches, an unset distribution mode and merge-on-read delete files all converge on the same outcome: thousands of small files, planning that reads more metadata than data, and tasks that spend their lives opening file handles. Compaction is a scheduled job, not an occasional cleanup, and it is invoked through a stored procedure on the catalog's system namespace.

CALL lake.system.rewrite_data_files(
  table       => 'sales.orders',
  strategy    => 'sort',
  sort_order  => 'customer_id, order_ts',
  where       => 'order_date < "2026-08-12"',
  options     => map(
    'min-input-files',          '5',
    'target-file-size-bytes',   '536870912',
    'partial-progress.enabled', 'true'
  )
);

The default strategy is binpack, which only repacks files to the target size without reordering rows — cheap, and enough to fix a small-file problem. sort additionally sorts within each file group, and a sort_order of the form zorder(col_a, col_b) interleaves two or more columns so that filters on either one prune. Sorting costs a shuffle; binpacking mostly does not.

Three arguments matter operationally. The where filter restricts the rewrite to a subset of partitions — always use it, both to bound the job and to keep compaction away from partitions an ingest job is currently writing. min-input-files stops the procedure from rewriting partitions that are already healthy. And partial-progress.enabled commits each file group as it completes instead of holding one enormous commit until the end; without it, a six-hour compaction that fails in hour five has accomplished nothing and has burned the cluster doing it.

Compaction is an ordinary writer and competes for the same optimistic commit. Rewriting a partition that concurrent ingest is appending to produces conflicts, and the compaction is the job that loses. Scoping it to closed partitions is not a nicety. On merge-on-read tables there is a second axis: the accumulated delete files themselves need consolidating, which recent Iceberg versions expose as a separate procedure — check what your version ships rather than assuming rewrite_data_files covers it.

Snapshot expiry, orphan files, and manifest rewriting

Compaction fixes the data files. Three further procedures keep the metadata and the storage bill bounded, and each has a distinct failure mode if you get the retention window wrong.

CALL lake.system.expire_snapshots(
  table => 'sales.orders', older_than => TIMESTAMP '2026-08-05 00:00:00', retain_last => 10);

CALL lake.system.remove_orphan_files(
  table => 'sales.orders', older_than => TIMESTAMP '2026-08-09 00:00:00');

CALL lake.system.rewrite_manifests(table => 'sales.orders');

Expiring snapshots is what actually deletes data. Until a snapshot expires, the files it referenced stay on storage — so the 51 GB your copy-on-write merge rewrote is still being paid for, twice, until expiry runs. Expiry also destroys time travel beyond the retention horizon and will break any incremental reader whose last-consumed snapshot has been removed. The table properties history.expire.max-snapshot-age-ms and history.expire.min-snapshots-to-keep supply the defaults when the arguments are omitted; set them deliberately, because the default horizon is days, not weeks.

Removing orphan files deletes files under the table location that no metadata references — the debris of jobs that died between writing data and committing. This is the most dangerous procedure Iceberg ships. It works by listing the entire table location, which is slow and expensive on object storage, and it cannot distinguish a file abandoned by a dead job from a file a live job is writing right now. The older_than guard is the only thing preventing the second case, which is why shortening it to make a cleanup finish faster is how tables get corrupted. Leave it at days and run the procedure rarely.

Rewriting manifests reorganises the metadata itself, clustering entries by partition so that planning opens fewer manifests. A table receiving frequent small commits accumulates a manifest per commit, and driver-side planning degrades long before scan performance does — a query with a multi-minute pause before its first task is usually this. Pair it with write.metadata.delete-after-commit.enabled and write.metadata.previous-versions-max so the chain of old metadata JSON files is trimmed on commit rather than growing forever.

A workable default schedule: compaction on hot partitions hourly or daily, manifest rewriting daily on high-commit tables, snapshot expiry daily with a retention window agreed with whoever depends on time travel, and orphan removal monthly with a generous older_than.

Getting existing tables in — migrate, snapshot, add_files

Three procedures convert an existing Hive table, and the difference between them is what happens to the original.

snapshot creates a new Iceberg table that references the source table's existing data files without copying or moving them, leaving the source completely intact. It is the test-drive: point your queries at the copy, confirm the plans and the numbers, and throw it away if you do not like what you see. Writes to the copy go to a new location so the source is never touched.

migrate converts the table in place. The original is renamed to a backup name and an Iceberg table takes over the identifier, so existing queries keep working against the same name. This requires SparkSessionCatalog, because the table being converted lives in the session catalog. Run snapshot first, always.

add_files registers Parquet or ORC files that already exist at some location into an Iceberg table you created yourself — the right tool when the source is not a Hive table at all, or when you want the new table's partitioning to differ from the old directory layout.

CALL lake.system.snapshot('spark_catalog.legacy.orders', 'lake.sales.orders_test');
CALL spark_catalog.system.migrate('legacy.orders');
CALL lake.system.add_files(table => 'sales.orders', source_table => '`parquet`.`s3://old-lake/orders`');

A note on the alternative you are probably also weighing. Delta Lake solves the same problem with a different shape: an ordered log of JSON actions in _delta_log with periodic checkpoints, addressable by path, maintained with OPTIMIZE, ZORDER and VACUUM. Iceberg puts a tree of metadata files behind a catalog pointer and exposes maintenance as CALL procedures. From inside a Spark job the day-to-day experience is similar; the divergence is elsewhere — Delta tables are reachable by path with no catalog at all, while Iceberg's catalog indirection is precisely what lets Flink and Trino commit to the same tables safely. Pick on the engine mix and the catalog you are willing to operate, not on the SQL syntax.

Failure modes and the operational playbook

The recurring ways a Spark and Iceberg deployment goes wrong, and what to do about each:

Extensions missing. Reads and appends work; the first CALL or ADD PARTITION FIELD fails with a parse error. Set spark.sql.extensions in the cluster defaults.

Hadoop catalog on object storage. Commits rely on atomic rename semantics that S3-style stores do not provide. Concurrent writers can lose commits. Move to a Hive, REST or Glue catalog before you have two writers.

Merge-on-read with no compaction. Reads get steadily slower as delete files accumulate and nobody can point at a change that caused it. Compaction is part of adopting merge-on-read, not a follow-up ticket.

Distribution mode left to chance. A partitioned write with no shuffle emits tasks × partitions files. Check count(*) and the average file size from the files metadata table after any new write job, not a week later.

Expiry that outruns its consumers. A retention window shorter than a downstream incremental reader's lag breaks that reader with an error about a missing snapshot. Agree the window with the consumers and encode it in table properties.

Orphan removal with a short window. Deletes files a running job is still writing. Days, not hours, and never during a heavy ingest window.

Slow planning on the driver. A long pause before the first task is metadata, not data. Look at the manifests metadata table, run rewrite_manifests, and bound the metadata history with the delete-after-commit properties.

Commit conflict storms. Several writers overwriting the same partitions will thrash the retry loop and then fail validation. Partition the workload so writers do not overlap, and reserve overwrite semantics for jobs that genuinely own their partitions; appends conflict with nothing and should be preferred wherever the data model allows.

The instrumentation for all of it is the same three queries: file count and average size from .files, the operation and summary columns from .snapshots, and manifest count from .manifests. Put them on a dashboard per table and most of the list above becomes visible while it is still cheap to fix.