Apache Iceberg: Table Format for the Data Lakehouse

1
2
3
4
5
6

-- Hive-style: users must write WHERE year=2026 AND month=3
-- or the query scans everything
SELECT * FROM hive.events WHERE year=2026 AND month=3 AND user_id = 42;

-- Iceberg: users write natural predicates, pruning is automatic
SELECT * FROM iceberg.events WHERE ts > TIMESTAMP '2026-03-01' AND user_id = 42;

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29

-- Partition by year, month, day extracted from a timestamp column
CREATE TABLE catalog.db.events (
  event_id  BIGINT,
  user_id   BIGINT,
  event_type STRING,
  ts        TIMESTAMP,
  payload   STRING
)
USING iceberg
PARTITIONED BY (year(ts), month(ts), day(ts));

-- Partition by bucket (for high-cardinality columns like user_id)
CREATE TABLE catalog.db.user_events (
  user_id   BIGINT,
  event_type STRING,
  ts        TIMESTAMP
)
USING iceberg
PARTITIONED BY (bucket(64, user_id), day(ts));

-- Partition by truncated string prefix
CREATE TABLE catalog.db.logs (
  host     STRING,
  level    STRING,
  message  STRING,
  ts       TIMESTAMP
)
USING iceberg
PARTITIONED BY (truncate(16, host), day(ts));

1
2
3
4

-- Change from daily to hourly partitioning going forward
ALTER TABLE catalog.db.events
REPLACE PARTITION FIELD day(ts)
WITH hour(ts);

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23

-- Add a new column (no data rewrite)
ALTER TABLE catalog.db.events
ADD COLUMN session_id STRING;

-- Drop a column (data retained in existing files, removed from schema)
ALTER TABLE catalog.db.events
DROP COLUMN payload;

-- Rename a column
ALTER TABLE catalog.db.events
RENAME COLUMN event_type TO event_name;

-- Reorder columns (metadata only)
ALTER TABLE catalog.db.events
ALTER COLUMN session_id FIRST;

-- Widen a numeric type (int → long, float → double)
ALTER TABLE catalog.db.events
ALTER COLUMN event_id TYPE BIGINT;

-- Add a nested column to a struct
ALTER TABLE catalog.db.events
ADD COLUMN metadata.source STRING;

1
2
3
4
5
6
7

-- View snapshot history
SELECT * FROM catalog.db.events.history
ORDER BY committed_at DESC;

-- View all snapshots with summary stats
SELECT snapshot_id, committed_at, operation, added_data_files_count, deleted_data_files_count
FROM catalog.db.events.snapshots;

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15

-- Query by snapshot ID
SELECT COUNT(*)
FROM catalog.db.events
FOR VERSION AS OF 8954597067493422955;

-- Query as of a specific timestamp
SELECT *
FROM catalog.db.events
FOR TIMESTAMP AS OF TIMESTAMP '2026-03-01 00:00:00 UTC';

-- Compare current vs yesterday
SELECT 'current' as version, COUNT(*) as cnt FROM catalog.db.events
UNION ALL
SELECT 'yesterday', COUNT(*) FROM catalog.db.events
FOR TIMESTAMP AS OF TIMESTAMP '2026-03-30 00:00:00 UTC';

1
2
3
4
5
6
7
8

-- Roll back the entire table to a previous snapshot (Spark)
CALL catalog.system.rollback_to_snapshot('db.events', 8954597067493422955);

-- Roll back to a timestamp (Trino)
CALL iceberg.system.rollback_to_timestamp(
  'db', 'events',
  TIMESTAMP '2026-03-30 12:00:00 UTC'
);

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14

# Create a feature branch
nessie branch feature/new-aggregations main

# All writes on this branch are isolated from main
spark.conf.set("spark.sql.catalog.nessie.ref", "feature/new-aggregations")

# Run experiments, transformations, schema changes
spark.sql("INSERT INTO nessie.db.agg_events SELECT ...")

# Merge back to main when validated
nessie merge feature/new-aggregations --into main

# Or discard the branch (no cleanup needed — data files aren't duplicated)
nessie delete branch feature/new-aggregations

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67

# docker-compose.yml
services:
  minio:
    image: minio/minio:latest
    ports:
      - "9000:9000"
      - "9001:9001"
    environment:
      MINIO_ROOT_USER: admin
      MINIO_ROOT_PASSWORD: password
    command: server /data --console-address ":9001"
    volumes:
      - minio-data:/data
    healthcheck:
      test: ["CMD", "mc", "ready", "local"]
      interval: 5s

  # Creates the warehouse bucket on startup
  minio-init:
    image: minio/mc:latest
    depends_on:
      minio:
        condition: service_healthy
    entrypoint: >
      /bin/sh -c "
      mc alias set local http://minio:9000 admin password;
      mc mb local/warehouse --ignore-existing;
      "

  # Iceberg REST catalog (backed by local filesystem for simplicity)
  iceberg-rest:
    image: apache/iceberg-rest-fixture:latest
    ports:
      - "8181:8181"
    environment:
      AWS_ACCESS_KEY_ID: admin
      AWS_SECRET_ACCESS_KEY: password
      AWS_REGION: us-east-1
      CATALOG_WAREHOUSE: s3://warehouse/
      CATALOG_IO__IMPL: org.apache.iceberg.aws.s3.S3FileIO
      CATALOG_S3_ENDPOINT: http://minio:9000
    depends_on:
      - minio-init

  # Spark with Iceberg + AWS S3 libraries
  spark:
    image: apache/spark:3.5.0-python3
    ports:
      - "4040:4040"   # Spark UI
      - "8888:8888"   # Jupyter
    environment:
      AWS_ACCESS_KEY_ID: admin
      AWS_SECRET_ACCESS_KEY: password
      AWS_REGION: us-east-1
    volumes:
      - ./notebooks:/opt/notebooks
    depends_on:
      - iceberg-rest
    command: >
      bash -c "
      pip install jupyter pyspark==3.5.0 pyiceberg &&
      jupyter notebook --ip=0.0.0.0 --port=8888 --no-browser
        --NotebookApp.token='' --notebook-dir=/opt/notebooks
      "

volumes:
  minio-data:

1
2
3
4
5

docker compose up -d

# MinIO console: http://localhost:9001 (admin/password)
# REST catalog:  http://localhost:8181
# Jupyter:       http://localhost:8888

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21

from pyspark.sql import SparkSession

spark = SparkSession.builder \
    .appName("iceberg-demo") \
    .config("spark.jars.packages",
        "org.apache.iceberg:iceberg-spark-runtime-3.5_2.12:1.5.2,"
        "software.amazon.awssdk:bundle:2.20.18,"
        "org.apache.hadoop:hadoop-aws:3.3.4") \
    .config("spark.sql.extensions",
        "org.apache.iceberg.spark.extensions.IcebergSparkSessionExtensions") \
    .config("spark.sql.catalog.demo", "org.apache.iceberg.spark.SparkCatalog") \
    .config("spark.sql.catalog.demo.type", "rest") \
    .config("spark.sql.catalog.demo.uri", "http://localhost:8181") \
    .config("spark.sql.catalog.demo.warehouse", "s3://warehouse/") \
    .config("spark.sql.catalog.demo.io-impl", "org.apache.iceberg.aws.s3.S3FileIO") \
    .config("spark.sql.catalog.demo.s3.endpoint", "http://localhost:9000") \
    .config("spark.hadoop.fs.s3a.endpoint", "http://localhost:9000") \
    .config("spark.hadoop.fs.s3a.access.key", "admin") \
    .config("spark.hadoop.fs.s3a.secret.key", "password") \
    .config("spark.hadoop.fs.s3a.path.style.access", "true") \
    .getOrCreate()

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39

from pyspark.sql.types import *
from datetime import datetime

# Create namespace
spark.sql("CREATE NAMESPACE IF NOT EXISTS demo.db")

# Create table with SQL
spark.sql("""
    CREATE TABLE IF NOT EXISTS demo.db.events (
        event_id  BIGINT,
        user_id   BIGINT,
        event_type STRING,
        ts        TIMESTAMP,
        props     MAP<STRING, STRING>
    )
    USING iceberg
    PARTITIONED BY (day(ts))
    TBLPROPERTIES (
        'format-version' = '2',
        'write.target-file-size-bytes' = '134217728'
    )
""")

# Insert via DataFrame API
from pyspark.sql import Row

rows = [
    Row(event_id=1, user_id=101, event_type="page_view",
        ts=datetime(2026, 3, 31, 10, 0, 0), props={"page": "/home"}),
    Row(event_id=2, user_id=102, event_type="purchase",
        ts=datetime(2026, 3, 31, 11, 30, 0), props={"amount": "99.99"}),
]

df = spark.createDataFrame(rows)
df.writeTo("demo.db.events").append()

# Or: overwrite a specific partition
df.writeTo("demo.db.events") \
    .overwritePartitions()

 1
 2
 3
 4
 5
 6
 7
 8
 9
10

-- Merge updates into the target table
MERGE INTO demo.db.users t
USING (
    SELECT 101 AS user_id, 'Alice Updated' AS name, 'alice@example.com' AS email
) s
ON t.user_id = s.user_id
WHEN MATCHED THEN
    UPDATE SET t.name = s.name, t.email = s.email
WHEN NOT MATCHED THEN
    INSERT (user_id, name, email) VALUES (s.user_id, s.name, s.email);

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14

# Files in the current snapshot
spark.sql("SELECT * FROM demo.db.events.files").show()

# Full snapshot history
spark.sql("SELECT * FROM demo.db.events.history").show()

# All snapshots with summary stats
spark.sql("SELECT * FROM demo.db.events.snapshots").show()

# Current partitions and row counts
spark.sql("SELECT * FROM demo.db.events.partitions").show()

# Manifest files
spark.sql("SELECT * FROM demo.db.events.manifests").show()

1
2
3
4
5
6
7
8
9

# etc/catalog/iceberg.properties
connector.name=iceberg
iceberg.catalog.type=rest
iceberg.rest-catalog.uri=http://iceberg-rest:8181
iceberg.file-format=PARQUET
hive.s3.endpoint=http://minio:9000
hive.s3.aws-access-key=admin
hive.s3.aws-secret-key=password
hive.s3.path-style-access=true

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34

-- Create table
CREATE TABLE iceberg.db.orders (
    order_id   BIGINT,
    customer   VARCHAR,
    total      DOUBLE,
    ordered_at TIMESTAMP WITH TIME ZONE
)
WITH (
    format = 'PARQUET',
    partitioning = ARRAY['day(ordered_at)', 'bucket(customer, 16)'],
    format_version = '2'
);

-- Insert
INSERT INTO iceberg.db.orders VALUES
    (1, 'alice', 49.99, TIMESTAMP '2026-03-31 10:00:00 UTC'),
    (2, 'bob',   129.99, TIMESTAMP '2026-03-31 11:00:00 UTC');

-- Time travel
SELECT * FROM iceberg.db.orders
FOR TIMESTAMP AS OF TIMESTAMP '2026-03-31 10:30:00 UTC';

-- Query metadata
SELECT * FROM iceberg.db.orders$snapshots;
SELECT * FROM iceberg.db.orders$history;
SELECT * FROM iceberg.db.orders$files;

-- Rollback
CALL iceberg.system.rollback_to_snapshot('db', 'orders', 1234567890);

-- Table maintenance
ALTER TABLE iceberg.db.orders EXECUTE expire_snapshots(retention_period => INTERVAL '7' DAY);
ALTER TABLE iceberg.db.orders EXECUTE remove_orphan_files();
ALTER TABLE iceberg.db.orders EXECUTE optimize;

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29

import duckdb

con = duckdb.connect()
con.install_extension("iceberg")
con.load_extension("iceberg")

# Read a local Iceberg table (useful for testing)
result = con.sql("""
    SELECT event_type, COUNT(*) as cnt
    FROM iceberg_scan('/tmp/warehouse/db/events', allow_moved_paths=true)
    GROUP BY event_type
    ORDER BY cnt DESC
""").fetchdf()
print(result)

# Read from S3
con.execute("""
    SET s3_endpoint='localhost:9000';
    SET s3_access_key_id='admin';
    SET s3_secret_access_key='password';
    SET s3_url_style='path';
    SET s3_use_ssl=false;
""")

result = con.sql("""
    SELECT * FROM iceberg_scan('s3://warehouse/db/events')
    WHERE ts > TIMESTAMP '2026-03-01'
    LIMIT 100
""").fetchdf()

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39

-- Flink SQL: create Iceberg catalog
CREATE CATALOG iceberg_catalog WITH (
  'type'            = 'iceberg',
  'catalog-type'    = 'rest',
  'uri'             = 'http://iceberg-rest:8181',
  'warehouse'       = 's3://warehouse',
  's3.endpoint'     = 'http://minio:9000',
  's3.access-key'   = 'admin',
  's3.secret-key'   = 'password',
  's3.path-style-access' = 'true'
);

-- Create target Iceberg table (format-version 2 required for upserts)
CREATE TABLE iceberg_catalog.db.user_events (
  user_id    BIGINT,
  event_type STRING,
  ts         TIMESTAMP(3),
  PRIMARY KEY (user_id) NOT ENFORCED
) WITH (
  'format-version' = '2',
  'write.upsert.enabled' = 'true'
);

-- Create Kafka source
CREATE TABLE kafka_events (
  user_id    BIGINT,
  event_type STRING,
  ts         TIMESTAMP(3),
  WATERMARK FOR ts AS ts - INTERVAL '5' SECOND
) WITH (
  'connector' = 'kafka',
  'topic'     = 'user-events',
  'properties.bootstrap.servers' = 'kafka:9092',
  'format'    = 'json'
);

-- Stream from Kafka into Iceberg with upsert
INSERT INTO iceberg_catalog.db.user_events
SELECT user_id, event_type, ts FROM kafka_events;

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17

-- Spark: rewrite small files into target size (128 MB default)
CALL demo.system.rewrite_data_files(
  table => 'db.events',
  strategy => 'binpack',
  options => map('target-file-size-bytes', '134217728',
                 'min-file-size-bytes', '67108864')
);

-- Rewrite only specific partitions
CALL demo.system.rewrite_data_files(
  table => 'db.events',
  where => 'ts >= TIMESTAMP ''2026-03-01'' AND ts < TIMESTAMP ''2026-04-01'''
);

-- Trino
ALTER TABLE iceberg.db.events EXECUTE optimize
WHERE ts >= DATE '2026-03-01';

 1
 2
 3
 4
 5
 6
 7
 8
 9
10

-- Spark: remove snapshots older than 7 days, keep at least 2
CALL demo.system.expire_snapshots(
  table => 'db.events',
  older_than => TIMESTAMP '2026-03-24 00:00:00',
  retain_last => 2
);

-- Trino
ALTER TABLE iceberg.db.events
EXECUTE expire_snapshots(retention_period => INTERVAL '7' DAY);

1
2
3
4
5
6
7
8

-- Spark: files in storage not referenced by any snapshot
CALL demo.system.remove_orphan_files(
  table => 'db.events',
  older_than => TIMESTAMP '2026-03-24 00:00:00'
);

-- Trino
ALTER TABLE iceberg.db.events EXECUTE remove_orphan_files();

1
2
3
4

# Airflow DAG or cron
# Daily: expire old snapshots + remove orphans
# Weekly: full compaction of last 7 days
# Monthly: compaction of older partitions (if write patterns were heavy)

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20

-- Source: Debezium CDC events from Kafka, landed in a staging table
-- Target: Production users table

MERGE INTO demo.db.users t
USING (
    -- Deduplicate: take latest operation per user_id
    SELECT user_id, name, email, address, op
    FROM (
        SELECT *,
               ROW_NUMBER() OVER (PARTITION BY user_id ORDER BY ts DESC) as rn
        FROM demo.db.users_cdc_staging
    ) WHERE rn = 1
) s
ON t.user_id = s.user_id
WHEN MATCHED AND s.op = 'DELETE' THEN DELETE
WHEN MATCHED THEN
    UPDATE SET t.name = s.name, t.email = s.email, t.address = s.address
WHEN NOT MATCHED AND s.op != 'DELETE' THEN
    INSERT (user_id, name, email, address)
    VALUES (s.user_id, s.name, s.email, s.address);

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16

# Read only new data since last run
last_snapshot = get_last_processed_snapshot()  # From your state store
current_snapshot = spark.sql(
    "SELECT snapshot_id FROM demo.db.events.snapshots ORDER BY committed_at DESC LIMIT 1"
).collect()[0][0]

incremental_df = spark.read \
    .option("start-snapshot-id", last_snapshot) \
    .option("end-snapshot-id", current_snapshot) \
    .table("demo.db.events")

# Process and save results
results = incremental_df.groupBy("event_type").count()
results.writeTo("demo.db.event_counts").overwritePartitions()

save_last_processed_snapshot(current_snapshot)

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20

# After a major migration, validate row counts match
before_count = spark.sql("""
    SELECT COUNT(*) as cnt
    FROM demo.db.events
    FOR TIMESTAMP AS OF TIMESTAMP '2026-03-30 23:59:59'
""").collect()[0]["cnt"]

after_count = spark.sql("SELECT COUNT(*) as cnt FROM demo.db.events").collect()[0]["cnt"]

assert after_count >= before_count, \
    f"Row count decreased: {before_count} → {after_count}"

# Find rows present before but missing after (should be 0 for append-only tables)
missing = spark.sql("""
    SELECT event_id FROM demo.db.events
    FOR TIMESTAMP AS OF TIMESTAMP '2026-03-30 23:59:59'
    EXCEPT
    SELECT event_id FROM demo.db.events
""")
assert missing.count() == 0, "Rows were deleted during migration"