Delta Lake & Apache Iceberg: Open Table Format Fundamentals

Open Table Formats: ACID for Data Lakes

Open table formats like Delta Lake and Apache Iceberg bring data warehouse capabilities to data lakes.

Why Open Table Formats Matter

Problems with Traditional Data Lakes:

• Partial writes corrupting data
• No transaction isolation
• Difficulty updating/deleting rows
• No time travel

How Open Table Formats Solve These Problems:

1. ACID transactions — reliable writes and updates
2. Time travel — query data at any point in time
3. Schema evolution — add/modify columns without breaking reads
4. Efficient data management — compaction, partition evolution

Key Insight: Open table formats add a metadata layer on top of Parquet files that enables ACID transactions, time travel, schema evolution, and efficient data management while maintaining open file formats.

Architecture Overview

Lakehouse Table Format Architecture

Delta vs Iceberg Comparison

Delta Lake

# Delta Lake: Create and write table
from delta import DeltaTable, SparkSessionBuilder
from pyspark.sql import SparkSession
from pyspark.sql.functions import *

spark = SparkSession.builder \
    .appName("DeltaLakeExamples") \
    .config("spark.sql.extensions", "io.delta.sql.DeltaSparkSessionExtension") \
    .config("spark.sql.catalog.spark_catalog", "org.apache.spark.sql.delta.catalog.DeltaCatalog") \
    .getOrCreate()

# Create Delta table
data = spark.range(0, 1000).withColumn("value", col("id") * 10)
data.write.format("delta").save("/delta/events")

# ACID Transaction: MERGE (Upsert)
from delta.tables import DeltaTable

delta_table = DeltaTable.forPath(spark, "/delta/events")

# Upsert new data
new_data = spark.range(500, 1500).withColumn("value", col("id") * 20)

delta_table.alias("target").merge(
    new_data.alias("source"),
    "target.id = source.id"
).whenMatchedUpdate(
    set={"value": "source.value"}
).whenNotMatchedInsert(
    values={"id": "source.id", "value": "source.value"}
).execute()

# Time Travel: Read previous version
previous_df = spark.read.format("delta").option("versionAsOf", 5).load("/delta/events")
current_df = spark.read.format("delta").load("/delta/events")

# Compare versions
diff = current_df.exceptAll(previous_df)
print(f"Rows added since version 5: {diff.count()}")

# Schema Evolution: Add new column
spark.sql("""
    ALTER TABLE delta.`/delta/events`
    ADD COLUMNS (event_type STRING, user_id BIGINT)
""")

# Schema Evolution: Rename column
spark.sql("""
    ALTER TABLE delta.`/delta/events`
    RENAME COLUMN value TO event_value
""")

# VACUUM: Clean up old files
delta_table.vacuum(retentionHours=168)  # 7 days

# Optimize: Compact small files
delta_table.optimize().executeCompaction()

# Z-Order: Optimize for specific columns
delta_table.optimize().executeZOrderBy("user_id", "event_type")

Apache Iceberg

# Iceberg: Create catalog and table
from pyspark.sql import SparkSession

spark = SparkSession.builder \
    .appName("IcebergExamples") \
    .config("spark.sql.catalog.iceberg", "org.apache.iceberg.spark.SparkCatalog") \
    .config("spark.sql.catalog.iceberg.type", "hadoop") \
    .config("spark.sql.catalog.iceberg.warehouse", "s3://iceberg-warehouse/") \
    .getOrCreate()

# Create Iceberg table with partitioning
spark.sql("""
    CREATE TABLE iceberg.analytics.events (
        event_id    BIGINT,
        event_type  STRING,
        user_id     BIGINT,
        event_time  TIMESTAMP,
        properties  STRING
    )
    USING iceberg
    PARTITIONED BY (days(event_time), event_type)
    LOCATION 's3://iceberg-warehouse/events/'
    TBLPROPERTIES (
        'format-version' = '2',
        'write.parquet.compression-codec' = 'snappy',
        'write.target-file-size-bytes' = '134217728',
        'commit.manifest-merge.enabled' = 'true'
    )
""")

# Insert data
spark.sql("""
    INSERT INTO iceberg.analytics.events
    SELECT
        id AS event_id,
        CASE WHEN id % 3 = 0 THEN 'click'
             WHEN id % 3 = 1 THEN 'view'
             ELSE 'purchase' END AS event_type,
        (id % 100) AS user_id,
        timestamp '2025-01-15 10:00:00' + interval (id % 86400) seconds AS event_time,
        CAST(id AS STRING) AS properties
    FROM range(10000)
""")

# Time Travel
spark.sql("""
    SELECT * FROM iceberg.analytics.events
    TIMESTAMP AS OF '2025-01-15 12:00:00'
""")

spark.sql("""
    SELECT * FROM iceberg.analytics.events
    VERSION AS OF 3
""")

# Partition Evolution (no data rewrite)
spark.sql("""
    ALTER TABLE iceberg.analytics.events
    ADD PARTITION FIELD event_type
""")
# Existing data remains in old partition spec
# New writes use the new spec

# Schema Evolution
spark.sql("""
    ALTER TABLE iceberg.analytics.events
    ADD COLUMNS (session_id BIGINT, page_url STRING)
""")

# Row-level delete
spark.sql("""
    DELETE FROM iceberg.analytics.events
    WHERE event_time < '2025-01-01'
""")

# Expire old snapshots
spark.sql("""
    CALL iceberg.system.expire_snapshots(
        table => 'analytics.events',
        older_than => TIMESTAMP '2025-01-01 00:00:00',
        retain_last => 10
    )
""")

Delta Lake vs. Iceberg Comparison

When to Choose Delta Lake vs. Iceberg

# Decision matrix implementation
class FormatSelector:
    """Select optimal table format based on requirements."""

    def __init__(self):
        self.scores = {"delta": 0, "iceberg": 0}

    def evaluate(self, requirements: dict) -> str:
        """Score each format against requirements."""
        if requirements.get("multi_engine"):
            self.scores["iceberg"] += 3

        if requirements.get("partition_evolution"):
            self.scores["iceberg"] += 2

        if requirements.get("spark_only"):
            self.scores["delta"] += 2

        if requirements.get("databricks"):
            self.scores["delta"] += 3

        if requirements.get("timestamp_travel"):
            self.scores["iceberg"] += 2

        if requirements.get("frequent_schema_changes"):
            self.scores["iceberg"] += 1

        return "iceberg" if self.scores["iceberg"] > self.scores["delta"] else "delta"

selector = FormatSelector()
recommendation = selector.evaluate({
    "multi_engine": True,
    "partition_evolution": True,
    "spark_only": False,
    "databricks": False,
    "timestamp_travel": True,
    "frequent_schema_changes": True
})
print(f"Recommended format: {recommendation}")

Key Concepts Summary

Performance Metrics

10 Best Practices

1. Use OPTIMIZE regularly on Delta Lake tables to compact small files (target: 128 MB per file)
2. Run VACUUM on Delta Lake after compaction to remove old files (default: 7-day retention)
3. Use Iceberg for multi-engine environments — better support for Spark, Trino, and Flink
4. Enable Z-Ordering on frequently filtered columns for data skipping
5. Partition by date and entity — partition pruning is the biggest performance lever
6. Set write.target-file-size-bytes to 128-256 MB for optimal file sizes
7. Expire old snapshots regularly to prevent metadata bloat
8. Use schema evolution carefully — test backward compatibility with consumers
9. Implement idempotent writes — use MERGE with unique keys to prevent duplicates
10. Monitor metadata size — large transaction logs slow down table reads

See Also

• Data Lake Architecture — Medallion architecture and schema-on-read patterns
• Data Lakehouse — Unified lake + warehouse with Databricks
• Partitioning & Indexing — Z-Ordering and bloom filter optimization
• Redshift Spectrum — Querying S3 data directly from Redshift
• Real-Time Analytics — Streaming ingestion into lakehouse tables
• MLOps for Data Engineering — Feature stores backed by Delta Lake