Apache Iceberg

Apache Iceberg is an open table format for huge analytic datasets. It brings SQL table semantics (ACID transactions, schema evolution, hidden partitioning, and time travel) to data lake files on S3/GCS/ADLS. In 2026, Iceberg has become the default open table format for the data lakehouse.

Table of Contents

1. What is Apache Iceberg?
2. Iceberg vs Delta Lake vs Hudi
3. Core Architecture
4. Getting Started
5. Core Operations
6. Schema Evolution
7. Partitioning & Hidden Partitioning
8. Time Travel & Snapshots
9. Table Maintenance
10. Iceberg Catalogs
11. Iceberg in 2026
12. Cheat Sheet

What is Apache Iceberg?

Before open table formats, data lakes had three options:

1. Raw Parquet/ORC files: fast reads, but no ACID, no schema, no history
2. Hive Metastore tables: basic metadata, but terrible at scale
3. Proprietary formats: vendor lock-in

Iceberg solves this by adding a metadata layer on top of existing file formats:

Your Data (Parquet/ORC/Avro files on S3)
           +
Iceberg Metadata Layer
           =
A "proper" SQL table with ACID, schema, history, and fast queries

Why Iceberg Won (vs Delta Lake vs Hudi)

Iceberg has become the industry standard because:

• Truly open: Apache Software Foundation, no single vendor owns it
• Catalog-agnostic: works with Hive Metastore, Glue, Nessie, REST, JDBC
• Engine-agnostic: Spark, Flink, Trino, Snowflake, Athena, DuckDB, BigQuery all read it
• Multi-writer support: designed for concurrent writes from multiple engines
• Column statistics: fine-grained stats for query pruning
• Partition evolution: change partition strategy without rewriting data

Iceberg vs Delta Lake vs Hudi

Core Architecture

┌─────────────────────────────────────────────────────────────────┐
│                        Iceberg Table                            │
├─────────────────────────────────────────────────────────────────┤
│ Catalog Layer   │  Hive / Glue / Nessie / REST / JDBC          │
│                 │  → stores: table location + current metadata  │
├─────────────────────────────────────────────────────────────────┤
│ Metadata Layer  │  metadata/v3.metadata.json                   │
│                 │  → schema, partition spec, snapshot history   │
├─────────────────────────────────────────────────────────────────┤
│ Manifest Layer  │  snap-123.avro  (manifest list)              │
│                 │  → list of manifest files                     │
│                 │  m1.avro, m2.avro  (manifests)               │
│                 │  → list of data files + column stats          │
├─────────────────────────────────────────────────────────────────┤
│ Data Layer      │  part-00001.parquet, part-00002.parquet, ... │
│                 │  → actual data (Parquet / ORC / Avro)         │
└─────────────────────────────────────────────────────────────────┘

Key Metadata Concepts

Getting Started

With PySpark

pip install pyspark pyiceberg

from pyspark.sql import SparkSession

spark = SparkSession.builder \
    .appName("Iceberg Demo") \
    .master("local[*]") \
    .config("spark.jars.packages",
            "org.apache.iceberg:iceberg-spark-runtime-3.5_2.12:1.6.0") \
    .config("spark.sql.extensions",
            "org.apache.iceberg.spark.extensions.IcebergSparkSessionExtensions") \
    .config("spark.sql.catalog.local", "org.apache.iceberg.spark.SparkCatalog") \
    .config("spark.sql.catalog.local.type", "hadoop") \
    .config("spark.sql.catalog.local.warehouse", "/tmp/iceberg-warehouse") \
    .getOrCreate()

With PyIceberg (no Spark needed for reads)

pip install "pyiceberg[s3fs,glue,duckdb]"

from pyiceberg.catalog import load_catalog

# Load from REST catalog
catalog = load_catalog("my_catalog", **{
    "type": "rest",
    "uri": "https://my-iceberg-catalog.com",
    "credential": "token:my-token",
})

# Load from AWS Glue
catalog = load_catalog("glue", **{
    "type": "glue",
    "region_name": "us-east-1",
})

# List tables
catalog.list_namespaces()
catalog.list_tables("my_db")

# Load a table
table = catalog.load_table("my_db.orders")

Core Operations

Create Table

-- SQL (via Spark)
CREATE TABLE local.db.orders (
    order_id    BIGINT,
    customer_id BIGINT,
    amount      DECIMAL(18, 2),
    status      STRING,
    created_at  TIMESTAMP
)
USING iceberg
PARTITIONED BY (days(created_at))    -- hidden partition!
TBLPROPERTIES (
    'write.format.default' = 'parquet',
    'write.parquet.compression-codec' = 'snappy'
);

# Python (PyIceberg)
from pyiceberg.schema import Schema
from pyiceberg.types import (
    NestedField, LongType, DecimalType, StringType, TimestampType
)
from pyiceberg.partitioning import PartitionSpec, PartitionField
from pyiceberg.transforms import DayTransform

schema = Schema(
    NestedField(1, "order_id", LongType(), required=True),
    NestedField(2, "customer_id", LongType()),
    NestedField(3, "amount", DecimalType(18, 2)),
    NestedField(4, "status", StringType()),
    NestedField(5, "created_at", TimestampType()),
)

partition_spec = PartitionSpec(
    PartitionField(source_id=5, field_id=1000, transform=DayTransform(), name="created_at_day")
)

table = catalog.create_table(
    identifier="my_db.orders",
    schema=schema,
    partition_spec=partition_spec,
    location="s3://my-bucket/iceberg/orders/",
)

Read and Write

# Write with Spark
df = spark.createDataFrame([
    (1, 100, 99.99, "completed", "2025-01-15 10:00:00"),
    (2, 101, 149.50, "pending", "2025-01-15 11:30:00"),
], ["order_id", "customer_id", "amount", "status", "created_at"])

df.writeTo("local.db.orders").append()
df.writeTo("local.db.orders").overwritePartitions()

# Read with Spark
orders = spark.table("local.db.orders")
orders.filter("status = 'completed'").show()

# Read with PyIceberg (no Spark!)
table = catalog.load_table("my_db.orders")
scan = table.scan(row_filter="status = 'completed'", limit=100)
df = scan.to_pandas()           # → Pandas DataFrame
arrow_table = scan.to_arrow()   # → PyArrow Table

Write Modes

# Append: add rows
df.writeTo("local.db.orders").append()

# Dynamic overwrite: replace only the partitions present in df
df.writeTo("local.db.orders").overwritePartitions()

# Overwrite rows matching a filter (keep it aligned with partition boundaries)
from pyspark.sql import functions as F
df.writeTo("local.db.orders").overwrite(
    (F.col("created_at") >= "2025-01-15") & (F.col("created_at") < "2025-01-16")
)

# Create or replace table
df.writeTo("local.db.orders").createOrReplace()

MERGE (Upsert)

MERGE INTO local.db.orders AS target
USING updates AS source
ON target.order_id = source.order_id
WHEN MATCHED AND source.status = 'cancelled' THEN DELETE
WHEN MATCHED THEN UPDATE SET
    target.amount = source.amount,
    target.status = source.status
WHEN NOT MATCHED THEN INSERT *;

# Python upsert (PyIceberg 0.9+); new_df is a PyArrow table
from pyiceberg.expressions import EqualTo

table = catalog.load_table("my_db.orders")
table.upsert(new_df, join_cols=["order_id"])

# Or replace every row matching a filter with new_df
table.overwrite(new_df, overwrite_filter=EqualTo("status", "pending"))

Schema Evolution

Iceberg's schema evolution is lossless: you can safely:

• Add columns (new reads get NULL for old rows)
• Drop columns (old files still have the data, just hidden)
• Rename columns: tracked by column ID, not name
• Promote types: int → long, float → double
• Reorder columns: cosmetic only

-- Add column
ALTER TABLE local.db.orders
ADD COLUMN discount DECIMAL(18, 2);

-- Rename column
ALTER TABLE local.db.orders
RENAME COLUMN discount TO discount_amount;

-- Drop column
ALTER TABLE local.db.orders
DROP COLUMN discount_amount;

-- Change type (widening only)
ALTER TABLE local.db.orders
ALTER COLUMN amount TYPE DECIMAL(24, 2);

-- Add nested column
ALTER TABLE local.db.orders
ADD COLUMN metadata.source STRING;

# Python schema evolution
from pyiceberg.types import DecimalType

table = catalog.load_table("my_db.orders")

with table.update_schema() as update:
    update.add_column("discount", DecimalType(18, 2))
    update.rename_column("status", "order_status")

Partitioning & Hidden Partitioning

Iceberg's hidden partitioning is a major advantage over Hive-style partitioning.

Hive Partitioning (old way: avoid)

-- Hive: partition column IS a real column
-- Users must write: WHERE dt = '2025-01-15'
-- Changing partition granularity requires rewriting ALL data
CREATE TABLE orders (order_id INT, amount FLOAT)
PARTITIONED BY (dt STRING);  -- user must manage this manually

Iceberg Hidden Partitioning (new way)

-- Iceberg: partition is derived from existing column
-- Iceberg automatically prunes the right partitions
-- Changing granularity is non-breaking!
CREATE TABLE orders (order_id INT, amount FLOAT, created_at TIMESTAMP)
USING iceberg
PARTITIONED BY (days(created_at));  -- hidden! User just queries on created_at

-- Query: no need to know about partitions
SELECT * FROM orders WHERE created_at BETWEEN '2025-01-01' AND '2025-01-31';
-- Iceberg automatically scans only January partition files

Partition Transforms

Partition Evolution

-- Change from day to hour partitioning: zero data rewrite!
ALTER TABLE local.db.orders
REPLACE PARTITION FIELD days(created_at)
WITH hours(created_at);

-- Old data stays in day partitions
-- New data goes into hour partitions
-- Queries work transparently across both

Time Travel & Snapshots

-- View snapshot history
SELECT * FROM local.db.orders.history;

-- Query at specific snapshot ID
SELECT * FROM local.db.orders
VERSION AS OF 4654808683286204734;

-- Query at specific timestamp
SELECT * FROM local.db.orders
TIMESTAMP AS OF '2025-01-15 10:00:00';

-- View all snapshots
SELECT * FROM local.db.orders.snapshots;

-- View data files for current snapshot
SELECT * FROM local.db.orders.files;

-- View manifests
SELECT * FROM local.db.orders.manifests;

# Python time travel
table = catalog.load_table("my_db.orders")

# List snapshots
for snapshot in table.history():
    print(snapshot.snapshot_id, snapshot.timestamp_ms)

# Scan at specific snapshot
scan = table.scan(snapshot_id=4654808683286204734)
df = scan.to_pandas()

Table Maintenance

-- Compact small files (bin-packing)
CALL local.system.rewrite_data_files('db.orders');

-- Compact with options
CALL local.system.rewrite_data_files(
    table => 'db.orders',
    strategy => 'sort',
    sort_order => 'zorder(customer_id, created_at)'
);

-- Remove old snapshots (expire)
CALL local.system.expire_snapshots(
    table => 'db.orders',
    older_than => TIMESTAMP '2025-01-01 00:00:00',
    retain_last => 5
);

-- Remove orphan files (files not referenced by any snapshot)
CALL local.system.remove_orphan_files(
    table => 'db.orders',
    older_than => TIMESTAMP '2025-01-01 00:00:00'
);

-- Rewrite manifests for faster planning
CALL local.system.rewrite_manifests('db.orders');

PyIceberg's maintenance support is newer and version-dependent (recent releases add a table.maintenance API for snapshot expiry), so check the docs for your version. Compaction and orphan-file cleanup are usually run through the Spark (or Flink) procedures above.

Iceberg Catalogs

A catalog tracks table locations and the current metadata pointer.

AWS Glue Catalog

spark = SparkSession.builder \
    .config("spark.sql.catalog.glue", "org.apache.iceberg.spark.SparkCatalog") \
    .config("spark.sql.catalog.glue.catalog-impl", "org.apache.iceberg.aws.glue.GlueCatalog") \
    .config("spark.sql.catalog.glue.warehouse", "s3://my-bucket/iceberg/") \
    .config("spark.sql.catalog.glue.io-impl", "org.apache.iceberg.aws.s3.S3FileIO") \
    .getOrCreate()

spark.sql("CREATE TABLE glue.mydb.orders (...) USING iceberg")

Nessie (Git for Data)

# Nessie lets you create branches of your data!
spark = SparkSession.builder \
    .config("spark.sql.catalog.nessie", "org.apache.iceberg.spark.SparkCatalog") \
    .config("spark.sql.catalog.nessie.catalog-impl", "org.apache.iceberg.nessie.NessieCatalog") \
    .config("spark.sql.catalog.nessie.uri", "http://localhost:19120/api/v2") \
    .config("spark.sql.catalog.nessie.ref", "main") \
    .getOrCreate()

# Work on a branch: isolates your changes
# (USE REFERENCE needs the Nessie Spark SQL extensions in spark.sql.extensions)
spark.sql("USE REFERENCE my_feature_branch IN nessie")
spark.sql("UPDATE nessie.db.orders SET status = 'v2' WHERE ...")

# Merge branch to main after validation
# (via Nessie API or CLI)

DuckDB + Iceberg

DuckDB can query Iceberg tables without Spark: perfect for local development and ad-hoc queries.

-- Install extension
INSTALL iceberg;
LOAD iceberg;

-- Query Iceberg table on S3
SELECT *
FROM iceberg_scan('s3://my-bucket/iceberg/orders/')
WHERE created_at >= '2025-01-01'
LIMIT 100;

-- Time travel
SELECT * FROM iceberg_scan(
    's3://my-bucket/iceberg/orders/',
    snapshot_from_id = 4654808683286204734
);

-- Via catalog (REST): the first argument is the warehouse name
ATTACH 'my_warehouse' AS iceberg_catalog (
    TYPE ICEBERG,
    ENDPOINT 'https://my-rest-catalog.com'
);
SELECT * FROM iceberg_catalog.my_db.orders LIMIT 10;

Iceberg in 2026

What Changed

Adoption in 2026

• AWS: S3 Tables (native Iceberg tables in S3), Athena + Glue
• GCP: BigQuery supports Iceberg as external tables
• Azure: ADLS + Iceberg via Fabric
• Snowflake: Iceberg tables in Snowflake (data lives in your S3)
• Databricks: Full Iceberg read/write support alongside Delta

Cheat Sheet

-- Create
CREATE TABLE catalog.db.tbl (col1 TYPE, ...) USING iceberg
PARTITIONED BY (days(ts_col));

-- Write
INSERT INTO catalog.db.tbl VALUES (...);
INSERT INTO catalog.db.tbl SELECT * FROM other_tbl;

-- Merge/Upsert
MERGE INTO target USING source ON condition
WHEN MATCHED THEN UPDATE SET ...
WHEN NOT MATCHED THEN INSERT *;

-- Time travel
SELECT * FROM tbl VERSION AS OF <snapshot_id>;
SELECT * FROM tbl TIMESTAMP AS OF '<timestamp>';

-- Schema
ALTER TABLE tbl ADD COLUMN new_col TYPE;
ALTER TABLE tbl RENAME COLUMN old TO new;
ALTER TABLE tbl DROP COLUMN col;

-- History / Snapshots
SELECT * FROM tbl.history;
SELECT * FROM tbl.snapshots;
SELECT * FROM tbl.files;
SELECT * FROM tbl.manifests;

-- Maintenance
CALL catalog.system.rewrite_data_files('db.tbl');
CALL catalog.system.expire_snapshots('db.tbl', TIMESTAMP '2025-01-01', 5);
CALL catalog.system.remove_orphan_files('db.tbl');

-- Partition evolution
ALTER TABLE tbl REPLACE PARTITION FIELD days(ts) WITH hours(ts);

# PyIceberg quick reference
from pyiceberg.catalog import load_catalog
from pyiceberg.types import StringType

catalog = load_catalog("my_catalog", type="rest", uri="https://...")
table = catalog.load_table("db.tbl")

# Scan
scan = table.scan(row_filter="col > 0", limit=100)
df = scan.to_pandas()

# Schema evolution
with table.update_schema() as upd:
    upd.add_column("new_col", StringType())

# Snapshots
for snap in table.history():
    print(snap.snapshot_id, snap.timestamp_ms)