Why Apache Iceberg is the Future of Data Lakehouses

-- Create an Iceberg table in AWS Glue Catalog
CREATE TABLE glue_catalog.analytics.events (
    event_id        STRING,
    user_id         STRING,
    event_type      STRING,
    event_timestamp TIMESTAMP,
    properties      MAP<STRING, STRING>,
    created_at      DATE
)
USING iceberg
PARTITIONED BY (days(event_timestamp), event_type)
LOCATION 's3://my-data-lake/analytics/events'
TBLPROPERTIES (
    'format-version' = '2',
    'write.metadata.compression-codec' = 'gzip'
);

-- Query data as it existed at a specific point in time
SELECT * FROM analytics.events
FOR SYSTEM_TIME AS OF TIMESTAMP '2025-01-01 00:00:00';

-- Query data at a specific snapshot
SELECT * FROM analytics.events
FOR SYSTEM_VERSION AS OF 1234567890;

-- View snapshot history
SELECT * FROM analytics.events.snapshots;

-- Rollback to a previous snapshot
CALL glue_catalog.system.rollback_to_snapshot(
    'analytics.events',
    1234567890
);

-- Add new columns (safe, backward compatible)
ALTER TABLE analytics.events
ADD COLUMNS (
    session_id STRING,
    device_type STRING
);

-- Rename columns (safe with Iceberg v2)
ALTER TABLE analytics.events
RENAME COLUMN properties TO event_properties;

-- Change column types (widening only)
ALTER TABLE analytics.events
ALTER COLUMN user_id TYPE BIGINT;

-- Drop columns (metadata only, no rewrite)
ALTER TABLE analytics.events
DROP COLUMN deprecated_field;

-- Original partitioning by day
CREATE TABLE events (...)
PARTITIONED BY (days(event_timestamp));

-- Evolve to hourly partitioning for recent data
-- (No data rewrite required!)
ALTER TABLE analytics.events
ADD PARTITION FIELD hours(event_timestamp);

-- Iceberg handles both partition schemes automatically
-- Old data: daily partitions
-- New data: hourly partitions
-- Queries work seamlessly across both

from pyspark.sql import SparkSession

# Configure Spark with Iceberg
spark = SparkSession.builder \
    .appName("IcebergAnalytics") \
    .config("spark.sql.catalog.glue_catalog", "org.apache.iceberg.spark.SparkCatalog") \
    .config("spark.sql.catalog.glue_catalog.warehouse", "s3://my-data-lake/") \
    .config("spark.sql.catalog.glue_catalog.catalog-impl", "org.apache.iceberg.aws.glue.GlueCatalog") \
    .config("spark.sql.extensions", "org.apache.iceberg.spark.extensions.IcebergSparkSessionExtensions") \
    .getOrCreate()

# Read Iceberg table
df = spark.read.format("iceberg").load("glue_catalog.analytics.events")

# Time travel query
df_historical = spark.read \
    .option("as-of-timestamp", "2025-01-01 00:00:00") \
    .format("iceberg") \
    .load("glue_catalog.analytics.events")

# Incremental read (changes since last snapshot)
df_changes = spark.read \
    .format("iceberg") \
    .option("start-snapshot-id", "123456") \
    .option("end-snapshot-id", "789012") \
    .load("glue_catalog.analytics.events")

-- Upsert pattern with MERGE INTO
MERGE INTO analytics.events AS target
USING staging.new_events AS source
ON target.event_id = source.event_id
WHEN MATCHED THEN
    UPDATE SET *
WHEN NOT MATCHED THEN
    INSERT *;

-- Delete with conditions
DELETE FROM analytics.events
WHERE event_timestamp < TIMESTAMP '2024-01-01'
  AND event_type = 'debug';

-- Update specific records
UPDATE analytics.events
SET properties = map('status', 'processed')
WHERE event_id = 'evt_123';

-- Expire old snapshots (keep last 5 days)
CALL glue_catalog.system.expire_snapshots(
    table => 'analytics.events',
    older_than => TIMESTAMP '2025-01-10 00:00:00',
    retain_last => 10
);

-- Remove orphan files
CALL glue_catalog.system.remove_orphan_files(
    table => 'analytics.events',
    older_than => TIMESTAMP '2025-01-01 00:00:00'
);

-- Compact small files
CALL glue_catalog.system.rewrite_data_files(
    table => 'analytics.events',
    options => map(
        'target-file-size-bytes', '134217728',  -- 128MB
        'min-input-files', '5'
    )
);

-- Rewrite manifests for faster planning
CALL glue_catalog.system.rewrite_manifests(
    'analytics.events'
);