apache-iceberg

Building data lakes with ACID transaction support and schema evolution

Apache Iceberg

Scope: Table format, schema/partition evolution, time travel, ACID transactions, catalog integration, performance optimization Lines: ~390 Last Updated: 2025-10-18

When to Use This Skill

Activate this skill when:

Core Concepts

Table Format Architecture

Three-layer metadata hierarchy:

ACID guarantees:

Hidden partitioning:

Catalog Integration

Catalog types:

Catalog configuration:

# PyIceberg with different catalogs
from pyiceberg.catalog import load_catalog

# Hive Metastore
catalog = load_catalog("hive",
    uri="thrift://localhost:9083",
    warehouse="s3://bucket/warehouse"
)

# AWS Glue
catalog = load_catalog("glue",
    warehouse="s3://bucket/warehouse"
)

# REST Catalog
catalog = load_catalog("rest",
    uri="https://catalog-api.example.com",
    credential="token:abc123"
)

Snapshots and Time Travel

Snapshot lifecycle:

Time travel queries:

Patterns

Table Creation with Schema and Partitioning

-- Spark SQL: Create Iceberg table with partitioning
CREATE TABLE catalog.db.events (
    event_id BIGINT,
    user_id STRING,
    event_type STRING,
    event_time TIMESTAMP,
    properties MAP<STRING, STRING>
)
USING iceberg
PARTITIONED BY (days(event_time), event_type)
TBLPROPERTIES (
    'write.format.default' = 'parquet',
    'write.parquet.compression-codec' = 'zstd',
    'write.metadata.compression-codec' = 'gzip'
);

Python PyIceberg:

from pyiceberg.catalog import load_catalog
from pyiceberg.schema import Schema
from pyiceberg.types import (
    NestedField, LongType, StringType, TimestampType, MapType
)
from pyiceberg.partitioning import PartitionSpec, PartitionField
from pyiceberg.transforms import DayTransform, IdentityTransform

catalog = load_catalog("default")

schema = Schema(
    NestedField(1, "event_id", LongType(), required=True),
    NestedField(2, "user_id", StringType(), required=True),
    NestedField(3, "event_type", StringType(), required=True),
    NestedField(4, "event_time", TimestampType(), required=True),
    NestedField(5, "properties", MapType(6, StringType(), 7, StringType()))
)

# Hidden partitioning: users query event_time, Iceberg handles partitioning
partition_spec = PartitionSpec(
    PartitionField(source_id=4, field_id=1000, transform=DayTransform(), name="event_day"),
    PartitionField(source_id=3, field_id=1001, transform=IdentityTransform(), name="event_type")
)

table = catalog.create_table(
    identifier="db.events",
    schema=schema,
    partition_spec=partition_spec,
    properties={"write.format.default": "parquet"}
)

When to use:

Schema Evolution

-- Add column (backward compatible)
ALTER TABLE catalog.db.events
ADD COLUMN session_id STRING;

-- Drop column (forward compatible)
ALTER TABLE catalog.db.events
DROP COLUMN properties;

-- Rename column (maintains column IDs)
ALTER TABLE catalog.db.events
RENAME COLUMN user_id TO customer_id;

-- Type promotion (safe widening)
ALTER TABLE catalog.db.events
ALTER COLUMN event_id TYPE BIGINT;  -- int -> bigint allowed

Python schema updates:

from pyiceberg.catalog import load_catalog
from pyiceberg.types import StringType, LongType

catalog = load_catalog("default")
table = catalog.load_table("db.events")

# Add column
with table.update_schema() as update:
    update.add_column("session_id", StringType())

# Rename column
with table.update_schema() as update:
    update.rename_column("user_id", "customer_id")

# Type promotion
with table.update_schema() as update:
    update.update_column("event_id", field_type=LongType())

Benefits:

Partition Evolution

-- Initial partitioning by day
CREATE TABLE catalog.db.logs (
    log_time TIMESTAMP,
    message STRING
)
USING iceberg
PARTITIONED BY (days(log_time));

-- Later: Add hourly partitioning for recent data
ALTER TABLE catalog.db.logs
REPLACE PARTITION FIELD days(log_time)
WITH hours(log_time);

-- Old data stays partitioned by day, new data by hour
-- Queries work seamlessly across both partition layouts

Partition transform types:

from pyiceberg.transforms import (
    YearTransform, MonthTransform, DayTransform, HourTransform,
    BucketTransform, TruncateTransform
)

# Date/time transforms
YearTransform()   # Extract year
MonthTransform()  # Extract month
DayTransform()    # Extract day
HourTransform()   # Extract hour

# Distribution transforms
BucketTransform(num_buckets=16)  # Hash bucket for uniform distribution
TruncateTransform(width=10)       # NOTE: Iceberg transform for trimming strings/numbers to width, not SQL TRUNCATE

When to use:

Time Travel and Snapshot Queries

-- Query as of specific timestamp
SELECT * FROM catalog.db.events
FOR SYSTEM_TIME AS OF TIMESTAMP '2025-10-01 00:00:00';

-- Query specific snapshot by ID
SELECT * FROM catalog.db.events
FOR SYSTEM_VERSION AS OF 1234567890;

-- View all snapshots
SELECT * FROM catalog.db.events.snapshots;

-- View snapshot history
SELECT
    snapshot_id,
    committed_at,
    operation,
    summary
FROM catalog.db.events.history
ORDER BY committed_at DESC;

-- Rollback to previous snapshot
CALL catalog.system.rollback_to_snapshot('db.events', 1234567890);

-- Cherry-pick specific snapshot
CALL catalog.system.cherrypick_snapshot('db.events', 1234567890);

Python time travel:

from datetime import datetime
from pyiceberg.catalog import load_catalog

catalog = load_catalog("default")
table = catalog.load_table("db.events")

# Read current snapshot
df = table.scan().to_arrow()

# Read specific snapshot
snapshot = table.snapshot_by_id(1234567890)
df_historical = table.scan(snapshot_id=snapshot.snapshot_id).to_arrow()

# Read as of timestamp
timestamp = datetime(2025, 10, 1, 0, 0, 0)
df_time_travel = table.scan(as_of_timestamp=timestamp).to_arrow()

# List all snapshots
for snapshot in table.snapshots():
    print(f"Snapshot {snapshot.snapshot_id} at {snapshot.timestamp_ms}")

When to use:

Incremental Reads and CDC

from pyiceberg.catalog import load_catalog
from pyiceberg.expressions import And, EqualTo, GreaterThanOrEqual

catalog = load_catalog("default")
table = catalog.load_table("db.events")

# Get current and previous snapshots
snapshots = list(table.snapshots())
current_snapshot = snapshots[-1]
previous_snapshot = snapshots[-2]

# Incremental read: only files added between snapshots
scan = table.scan(
    snapshot_id=current_snapshot.snapshot_id,
    from_snapshot_id=previous_snapshot.snapshot_id
)

# Filter incremental data
incremental_df = scan.filter(
    GreaterThanOrEqual("event_time", "2025-10-18 00:00:00")
).to_arrow()

print(f"Incremental records: {len(incremental_df)}")

Spark incremental reads:

// Read changes between two snapshots
val incrementalDF = spark.read
  .format("iceberg")
  .option("start-snapshot-id", "1234567890")
  .option("end-snapshot-id", "1234567891")
  .load("catalog.db.events")

// Read all changes since timestamp
val cdcDF = spark.read
  .format("iceberg")
  .option("start-timestamp", "2025-10-18 00:00:00")
  .load("catalog.db.events")

When to use:

ACID Transactions with Spark

import org.apache.spark.sql.SparkSession
import org.apache.iceberg.spark.extensions.IcebergSparkSessionExtensions

val spark = SparkSession.builder()
  .appName("Iceberg ACID")
  .config("spark.sql.extensions",
    "org.apache.iceberg.spark.extensions.IcebergSparkSessionExtensions")
  .config("spark.sql.catalog.catalog", "org.apache.iceberg.spark.SparkCatalog")
  .config("spark.sql.catalog.catalog.type", "hive")
  .getOrCreate()

// Atomic write: all or nothing
val newEvents = spark.read.json("s3://bucket/raw/events/")
newEvents.writeTo("catalog.db.events")
  .append()  // Atomic commit

// Upsert with MERGE INTO (atomic)
spark.sql("""
  MERGE INTO catalog.db.events t
  USING updates s
  ON t.event_id = s.event_id
  WHEN MATCHED THEN UPDATE SET *
  WHEN NOT MATCHED THEN INSERT *
""")

// Delete with predicate (atomic)
spark.sql("""
  DELETE FROM catalog.db.events
  WHERE event_time < TIMESTAMP '2024-01-01'
""")

// Multi-table transaction (via WAP - Write-Audit-Publish)
spark.conf.set("spark.wap.id", "staging-branch")

// Write to staging branch
df.writeTo("catalog.db.events").append()

// Validate data
val count = spark.sql("SELECT COUNT(*) FROM catalog.db.events").collect()

// Publish if valid (atomic promotion)
if (count.head.getLong(0) > 0) {
  spark.sql("CALL catalog.system.cherrypick_snapshot('db.events', 'staging-branch')")
}

When to use:

Table Maintenance and Optimization

-- Expire old snapshots (free storage)
CALL catalog.system.expire_snapshots(
  table => 'db.events',
  older_than => TIMESTAMP '2025-09-01 00:00:00',
  retain_last => 10
);

-- Remove orphan files (unreferenced data files)
CALL catalog.system.remove_orphan_files(
  table => 'db.events',
  older_than => TIMESTAMP '2025-10-01 00:00:00'
);

-- Rewrite small files into larger files
CALL catalog.system.rewrite_data_files(
  table => 'db.events',
  options => map('target-file-size-bytes', '536870912')  -- 512 MB
);

-- Rewrite manifest files for faster planning
CALL catalog.system.rewrite_manifests('db.events');

-- Compact data files in partition
CALL catalog.system.rewrite_data_files(
  table => 'db.events',
  where => 'event_date = DATE "2025-10-18"'
);

Python maintenance:

from pyiceberg.catalog import load_catalog
from datetime import datetime, timedelta

catalog = load_catalog("default")
table = catalog.load_table("db.events")

# Expire snapshots older than 30 days
expire_before = datetime.now() - timedelta(days=30)
table.expire_snapshots(
    older_than=expire_before,
    retain_last=10  # Keep at least 10 snapshots
)

# Get table statistics
metadata = table.metadata
print(f"Snapshots: {len(metadata.snapshots)}")
print(f"Schema version: {metadata.current_schema_id}")
print(f"Partition spec: {metadata.default_spec_id}")

When to use:

Integration with Flink for Streaming

import org.apache.flink.table.api.TableEnvironment;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
TableEnvironment tEnv = TableEnvironment.create(env);

// Configure Iceberg catalog
tEnv.executeSql(
  "CREATE CATALOG iceberg_catalog WITH (" +
  "  'type'='iceberg'," +
  "  'catalog-type'='hive'," +
  "  'uri'='thrift://localhost:9083'," +
  "  'warehouse'='s3://bucket/warehouse'" +
  ")"
);

tEnv.useCatalog("iceberg_catalog");

// Create Iceberg table from Flink
tEnv.executeSql(
  "CREATE TABLE events (" +
  "  event_id BIGINT," +
  "  user_id STRING," +
  "  event_time TIMESTAMP(6)," +
  "  PRIMARY KEY (event_id) NOT ENFORCED" +
  ") PARTITIONED BY (event_type)" +
  " WITH ('write.format.default'='parquet')"
);

// Stream write to Iceberg (micro-batch commits)
tEnv.executeSql(
  "INSERT INTO events " +
  "SELECT event_id, user_id, event_time FROM kafka_source"
);

// Stream read from Iceberg
tEnv.executeSql(
  "SELECT * FROM events " +
  "/*+ OPTIONS('streaming'='true', 'monitor-interval'='10s') */"
);

When to use:

Quick Reference

Core Commands

| Operation | Spark SQL | Python PyIceberg | |-----------|-----------|------------------| | Create table | CREATE TABLE ... USING iceberg | catalog.create_table() | | Add column | ALTER TABLE ... ADD COLUMN | table.update_schema().add_column() | | Rename column | ALTER TABLE ... RENAME COLUMN | table.update_schema().rename_column() | | Time travel | FOR SYSTEM_TIME AS OF | table.scan(as_of_timestamp=...) | | Expire snapshots | CALL system.expire_snapshots() | table.expire_snapshots() | | Rewrite files | CALL system.rewrite_data_files() | N/A (use Spark) |

Partition Transforms

Transform         | Example                    | Use Case
------------------|----------------------------|-----------------------
years(ts)         | PARTITIONED BY (years(ts)) | Long-term archives
months(ts)        | PARTITIONED BY (months(ts))| Monthly aggregations
days(ts)          | PARTITIONED BY (days(ts))  | Daily pipelines
hours(ts)         | PARTITIONED BY (hours(ts)) | Streaming ingestion
bucket(N, id)     | PARTITIONED BY (bucket(16, id)) | Uniform distribution
truncate(W, str)  | PARTITIONED BY (truncate(10, str)) | String prefixes  <!-- NOTE: Iceberg transform function -->

Configuration Best Practices

# Write performance
write.format.default=parquet
write.parquet.compression-codec=zstd
write.target-file-size-bytes=536870912  # 512 MB

# Metadata optimization
write.metadata.compression-codec=gzip
write.metadata.metrics.default=truncate(16)  # NOTE: Iceberg's truncate transform for metrics, not SQL TRUNCATE

# Commit behavior
commit.retry.num-retries=4
commit.retry.min-wait-ms=100

# Snapshot expiration
history.expire.max-snapshot-age-ms=432000000  # 5 days

Catalog URLs

Catalog Type | URI Format                              | Example
-------------|------------------------------------------|------------------
Hive         | thrift://host:port                      | thrift://localhost:9083
Glue         | (use AWS config)                        | N/A
REST         | https://catalog-host/path               | https://api.example.com
JDBC         | jdbc:postgresql://host:port/db          | jdbc:postgresql://localhost:5432/iceberg
Nessie       | https://nessie-host/api/v1              | https://nessie.example.com/api/v1

Anti-Patterns

❌ Partitioning by high-cardinality columns: Creating too many partitions (user_id, session_id) ✅ Use bucketing for high-cardinality: PARTITIONED BY (bucket(256, user_id))

❌ Manually managing partition columns in queries: WHERE date_partition = '2025-10-18' ✅ Use hidden partitioning: WHERE event_time >= '2025-10-18' (Iceberg handles partitions)

❌ Never expiring snapshots: Unbounded metadata growth, slow query planning ✅ Schedule regular snapshot expiration: CALL system.expire_snapshots() weekly

❌ Creating many small files: Thousands of tiny files per partition ✅ Rewrite to target file size (512 MB - 1 GB): CALL system.rewrite_data_files()

❌ Using Hive table format for schema evolution: Requires full table rewrites ✅ Use Iceberg's built-in schema evolution: ALTER TABLE ... ADD COLUMN

❌ Querying without partition filters: Full table scans on large datasets ✅ Always filter on partition columns: WHERE event_time BETWEEN ... AND ...

❌ Ignoring snapshot metadata: Missing opportunities for incremental processing ✅ Use snapshot IDs for incremental reads: scan(from_snapshot_id=...)

❌ Mixing partition schemes without evolution: Manual partition column changes ✅ Use partition evolution: REPLACE PARTITION FIELD ... WITH ...

Related Skills

Last Updated: 2025-10-18 Format Version: 1.0 (Atomic)