Apache Iceberg: An Engine-Agnostic Table Format for the Lakehouse

Apache Iceberg is an open table format that turns a directory of Parquet (or ORC/Avro) files on object storage into a real table with snapshots, schema evolution, and safe concurrent writes. What sets it apart from other lakehouse formats is its catalog-centric design and hidden partitioning: the table is defined by a catalog pointer to a metadata tree, and partitioning is a property of the table rather than something queries must spell out. This tutorial works through Iceberg from first principles using PyIceberg as the primary hands-on path, then shows the same table from Spark SQL and notes how DuckDB and Trino read it.

What Iceberg Is and Why a Table Format Matters

A plain data lake is just files in a bucket. Parquet gives you efficient columnar storage, but a folder of Parquet files is not a table: there is no atomic commit, no agreed-upon schema, no way to know which files are current, and no safe way for two jobs to write at once. A reader scanning the directory while a writer is mid-flight can pick up half-written data.

A table format solves this by adding a layer of metadata on top of the data files. That metadata answers three questions precisely:

• Which files make up the table right now (and at any past point in time)?
• What is the schema, and how has it changed?
• How is the data laid out so a query can skip files it does not need?

Iceberg's answer is a tree of immutable metadata files plus a catalog that holds a single pointer to the current top of that tree. Swapping that pointer atomically is what makes a commit. Because every metadata file is immutable, readers always see a consistent snapshot and never observe a partial write.

How Iceberg Differs From Delta Lake and Hudi

All three are open table formats over columnar files, but they make different design choices:

• Iceberg tracks state through a catalog pointer and a tree of metadata/manifest files. Partitioning is hidden — declared on the table and applied automatically — and Iceberg is explicitly engine-agnostic, with a stable spec implemented by Spark, Flink, Trino, DuckDB, Snowflake, and the pure-Python PyIceberg.
• Delta Lake tracks state through an ordered JSON transaction log (deltalog) co-located with the data. Partitioning is by physical directory layout, and the ecosystem historically centered on Spark (though delta-rs broadened it).
• Hudi focuses on fast upserts and incremental streams, with a timeline of commits and a strong record-key/primary-key model. It is well suited to CDC-heavy workloads.

The practical takeaway: choose Iceberg when you want multiple engines to read and write the same table through a shared catalog, and when you want partitioning and schema changes to be metadata operations rather than data rewrites.

Iceberg Architecture

An Iceberg table is a layered tree. From the top down:

Catalog                          (points to the current metadata file per table)
  └─ metadata.json               (table schema, partition specs, snapshot list, current snapshot id)
       └─ manifest list          (one Avro file per snapshot; lists the manifests in that snapshot)
            └─ manifest files     (Avro; each lists data files + per-file stats and partition values)
                 └─ data files     (Parquet / ORC / Avro holding the actual rows)

Reading top to bottom:

• The catalog maps a table identifier (e.g. analytics.events) to the location of its current metadata.json. This is the only mutable pointer; everything below is immutable.
• The metadata file describes the table: its current schema, all historical schemas, the partition specs, sort orders, table properties, and the list of snapshots. A commit writes a new metadata file and atomically updates the catalog pointer to it.
• A snapshot is a complete, point-in-time view of the table. Each snapshot references one manifest list.
• The manifest list enumerates the manifest files that belong to the snapshot, with summary statistics (row counts, partition value ranges) so a planner can prune entire manifests.
• Each manifest file lists actual data files along with per-file column statistics (min/max, null counts) and the file's partition values. These stats drive file-level pruning.

Because pruning happens by reading compact metadata rather than opening data files, query planning scales to millions of files. And because each snapshot is a self-contained tree of immutable files, time travel is simply reading an older snapshot's manifest list.

Catalogs: REST, SQL/JDBC, Glue, Hive

The catalog is where Iceberg differs most visibly from a log-based format. Common implementations:

• REST catalog — a service speaking the Iceberg REST protocol. The modern, engine-neutral default for production; backends include Nessie, Polaris, Lakekeeper, and managed services.
• SQL / JDBC catalog — stores table pointers in a relational database (SQLite, Postgres, MySQL). Excellent for local development and small deployments. We use a local SQLite SQL catalog in this tutorial.
• AWS Glue catalog — uses the Glue Data Catalog as the metastore; native on AWS.
• Hive catalog — uses a Hive Metastore, common in existing Hadoop/Spark estates.

The same physical data files can be registered in different catalogs; the catalog only governs the pointer and namespace/table registration.

PyIceberg: The Hands-On Path

PyIceberg is a pure-Python implementation that reads and writes Iceberg tables without a JVM or Spark. It is ideal for notebooks, single-node ETL, and embedding Iceberg into pandas/PyArrow pipelines.

Installation

pip install "pyiceberg[sql-sqlite,pyarrow]" pandas

This pulls in PyIceberg with the SQLite-backed SQL catalog and the PyArrow data layer. Add extras like s3fs or glue when you move beyond local storage.

Creating a Local SQL Catalog

We will model an events table for a product analytics use case. First, create a catalog backed by a local SQLite database, with data files written under a local warehouse directory.

import os
from pyiceberg.catalog.sql import SqlCatalog

warehousepath = os.path.abspath("warehouse")

os.makedirs(warehousepath, existok=True)


catalog = SqlCatalog(
    "local",
    {

        "uri": f"sqlite:///{warehousepath}/catalog.db",
        "warehouse": f"file://{warehousepath}",

    },
)

from pyiceberg.schema import Schema
from pyiceberg.types import (
    NestedField, StringType, LongType, TimestampType, DoubleType,
)

catalog.createnamespaceifnotexists("analytics")


schema = Schema(
    NestedField(fieldid=1, name="eventid", fieldtype=LongType(), required=True),
    NestedField(fieldid=2, name="userid", fieldtype=LongType(), required=True),

    NestedField(fieldid=3, name="eventtype", fieldtype=StringType(), required=True),
    NestedField(fieldid=4, name="eventts", fieldtype=TimestampType(), required=True),

    NestedField(fieldid=5, name="amount", fieldtype=DoubleType(), required=False),

)

table = catalog.createtableifnotexists(

    identifier="analytics.events",
    schema=schema,
)

import pyarrow as pa
from datetime import datetime

rows = pa.table(
    {
        "eventid": [1, 2, 3, 4],

        "userid": [1001, 1002, 1001, 1003],
        "eventtype": ["view", "click", "purchase", "view"],

        "eventts": [
            datetime(2026, 5, 1, 9, 0),
            datetime(2026, 5, 1, 9, 5),
            datetime(2026, 5, 2, 14, 30),
            datetime(2026, 5, 3, 11, 15),
        ],
        "amount": [None, None, 49.90, None],
    },
    schema=table.schema().asarrow(),

)

table.append(rows)

import pandas as pd

pdf = pd.DataFrame(
    {
        "eventid": [5, 6],

        "userid": [1002, 1004],
        "eventtype": ["click", "purchase"],

        "eventts": pd.todatetime(["2026-05-04 08:00", "2026-05-04 08:30"]),

        "amount": [None, 19.99],
    }
)
table.append(pa.Table.frompandas(pdf, schema=table.schema().asarrow()))

df = (
    table.scan(
        rowfilter="eventtype == 'purchase'",

        selectedfields=("eventid", "userid", "amount", "eventts"),

    )
    .topandas()
)
print(df)

from pyiceberg.expressions import And, GreaterThanOrEqual, EqualTo
from datetime import datetime

scan = table.scan(
    rowfilter=And(

        EqualTo("eventtype", "view"),
        GreaterThanOrEqual("eventts", datetime(2026, 5, 2)),

    ),
)
arrowtbl = scan.toarrow()   # zero-copy Arrow; or .topandas(), .toduckdb("events")

with table.updateschema() as upd:
    upd.addcolumn("sessionid", StringType(), doc="Client session identifier")

    upd.renamecolumn("amount", "amountusd")

    upd.updatecolumn("userid", required=False)


Dropping is equally safe; old data files are simply not read for that column id.
with table.updateschema() as upd:
    upd.deletecolumn("sessionid")

• Add appends a new column with a fresh field ID; existing rows read back as null.
• Rename changes only the name attached to an ID; data files are untouched.
• Drop stops projecting a column ID; no data rewrite is required.
• Reorder* and type promotions (e.g. int → long) are supported without rewriting files.

from pyiceberg.transforms import DayTransform, BucketTransform

with table.updatespec() as spec:

    spec.addfield("eventts", DayTransform(), partitionfieldname="eventday")
    spec.addfield("userid", BucketTransform(numbuckets=8), partitionfieldname="userbucket")

for entry in table.history():
    print(entry.snapshotid, entry.madecurrentat)

for snap in table.snapshots():
    print(snap.snapshotid, snap.summary.operation, snap.summary)

oldsnapshotid = table.history()[0].snapshotid
dfold = table.scan(snapshotid=oldsnapshotid).topandas()

table.managesnapshots().rollbacktosnapshot(oldsnapshotid).commit()

from pyiceberg.expressions import EqualTo

Delete all 'view' events, then optionally append replacements.
table.delete(deletefilter=EqualTo("eventtype", "view"))

Overwrite rows matching a predicate with new data in one commit:
table.overwrite(
    df=correctedrows,                      # a PyArrow table

    overwritefilter=EqualTo("eventtype", "click"),

)

# Requires identifier (key) columns; updates matches, inserts the rest.
result = table.upsert(
    df=incomingrows,                       # PyArrow table with same schema
    joincols=["eventid"],
)
print(result.rowsupdated, result.rowsinserted)

• Copy-on-write (CoW) rewrites every data file that contains an affected row at write time. Reads stay fast (no merge needed), but writes are heavier. Good for read-heavy tables with infrequent updates.
• Merge-on-read (MoR) writes lightweight delete files (position or equality deletes) alongside data, and readers merge them on the fly. Writes are cheap and fast — ideal for streaming/CDC — at the cost of more work at read time until compaction folds the deletes back in.

from datetime import datetime, timedelta

cutoffms = int((datetime.now() - timedelta(days=7)).timestamp()  1000)
table.expiresnapshots().expiresnapshotsolderthan(cutoffms).commit()

This both frees storage and trims the metadata that planners must read.

Compaction / Rewrite

Frequent small appends produce many small files, which hurts scan performance. Compaction rewrites them into fewer, larger files (and can re-sort or re-cluster by the partition spec). PyIceberg is adding native rewrite support; in production this is commonly run from Spark:

CALL local.system.rewritedatafiles(
  table => 'analytics.events',
  options => map('target-file-size-bytes', '134217728')
);

Orphan File Cleanup

Failed writes can leave data files that no snapshot references. After expiring snapshots, remove orphans (typically from Spark, since it scans the storage layer):

CALL local.system.removeorphanfiles(table => 'analytics.events');

Run orphan cleanup conservatively (with an age threshold) so it never deletes files a concurrent write is still committing.

Querying the Same Table From Spark SQL

Because the table lives in a shared catalog, other engines read and write it without conversion. A Spark session configured against the same SQL/JDBC or REST catalog sees analytics.events directly.

-- Create or reference an Iceberg table from Spark SQL
CREATE TABLE local.analytics.events (
    eventid    BIGINT,

    userid     BIGINT,
    eventtype  STRING,

    eventts    TIMESTAMP,
    amountusd  DOUBLE

)
USING iceberg
PARTITIONED BY (days(eventts), bucket(8, userid));


-- Merge (upsert) incoming CDC rows
MERGE INTO local.analytics.events t
USING stagedevents s
ON t.eventid = s.eventid
WHEN MATCHED THEN UPDATE SET 

WHEN NOT MATCHED THEN INSERT ;

-- Time travel by snapshot version or timestamp
SELECT eventtype, count() AS n

FROM local.analytics.events VERSION AS OF 3623387065522815048
GROUP BY eventtype;

SELECT  FROM local.analytics.events TIMESTAMP AS OF '2026-05-02 00:00:00';

The PARTITIONED BY (days(eventts), bucket(8, userid)) clause is hidden partitioning in SQL form — note that queries still filter on eventts and userid directly.

Reading Iceberg From DuckDB and Trino

• DuckDB reads Iceberg through its iceberg extension (INSTALL iceberg; LOAD iceberg;) and can attach a REST catalog, making it a fast single-node query engine over lakehouse tables. PyIceberg also offers scan.toduckdb("events") to register a scan as a DuckDB relation.
• Trino treats Iceberg as a first-class connector: point its iceberg catalog at the same metastore (Glue, Hive, REST) and run full SQL, including INSERT, MERGE, time travel, and the same maintenance procedures.

• Use a managed/REST catalog in production. Local SQLite is fine for development, but a REST catalog (Nessie, Polaris, Lakekeeper) gives you safe multi-engine concurrency and access control.
• Partition by access pattern, not cardinality. Use day(ts) or month(ts) for time-series reads; use bucket(N, key) for high-cardinality join/filter keys. Avoid over-partitioning into tiny files.
• Push filters and projections into scan(...). Always pass rowfilter and selected_fields so manifest/file statistics can prune work early.
• Schedule maintenance. Regularly expire snapshots, compact small files, and clean orphans. These three jobs prevent the slow degradation that plagues unmaintained lake tables.
• Pick CoW vs MoR per workload. Read-heavy dimension tables favor copy-on-write; streaming/CDC fact tables favor merge-on-read with periodic compaction.
• Lean on partition and schema evolution. Change layout as access patterns shift without rewriting history; let stable field IDs make renames safe.
• Keep retention and compliance in mind. Time travel keeps old data reachable, so align snapshot expiration with your data deletion obligations.

Conclusion and Key Takeaways

Apache Iceberg makes a bucket of Parquet files behave like a real, transactional table while keeping it open to any engine. Its defining traits are a catalog pointer over an immutable metadata tree, hidden partitioning driven by transforms, ID-based schema evolution, and a deliberate engine-agnostic spec.

• A table is a catalog pointer → metadata.json → manifest list → manifests → data files; commits atomically swap the pointer.
• PyIceberg gives you a full read/write path in pure Python: create a SQL catalog, define a Schema, append Arrow/pandas data, and scan(...) with filters and projection.
• Hidden partitioning lets you filter on natural columns while Iceberg prunes via transforms (day, month, bucket, truncate), and supports partition evolution without rewrites.
• Time travel, rollback, deletes, and upserts are all snapshot operations; choose copy-on-write for read-heavy tables and merge-on-read for streaming/CDC.
• Maintenance — expiring snapshots, compaction, and orphan cleanup — keeps tables fast and storage costs bounded.
• The same table is readable from Spark, Trino, and DuckDB through a shared catalog, which is exactly what distinguishes Iceberg from log-centric formats like Delta Lake and upsert-focused ones like Hudi.

Start with a local SQL catalog and PyIceberg to learn the model, then graduate to a REST catalog and a distributed engine when you need multi-engine concurrency at scale.