Partition Evolution — Uniek voor Iceberg

Dit is de feature die Iceberg onderscheidt. Bij Delta Lake en Hudi moet je bij een partitioneringswijziging de volledige tabel herschrijven. Bij Iceberg kun je de partitioneringsstrategie aanpassen zonder één byte data te verplaatsen.

-- Begin met partitionering per dag
CREATE TABLE orders (
  order_id BIGINT,
  customer_id BIGINT,
  amount DECIMAL(10,2),
  order_date TIMESTAMP
)
USING iceberg
PARTITIONED BY (days(order_date));

-- Later groeien naar per uur (zonder tabel te herschrijven!)
ALTER TABLE orders
ADD PARTITION FIELD hours(order_date);

-- Iceberg schrijft nieuwe data in de nieuwe partitiestructuur,
-- oude data blijft ongewijzigd. Query engines begrijpen beide.

Time Travel & Rollback

Query historische data of herstel per ongeluk gewiste records:

-- Query de tabel van gisteren (Spark SQL)
SELECT * FROM orders
TIMESTAMP AS OF '2026-03-20 09:00:00';

-- Query specifieke snapshot
SELECT * FROM orders VERSION AS OF 4521;

-- Rollback naar vorige snapshot
CALL catalog.system.rollback_to_snapshot('db.orders', 4521);

-- Bekijk snapshot history
SELECT * FROM orders.snapshots;

Schema Evolution zonder Downtime

Voeg kolommen toe, hernoem of verwijder ze — veilig en zonder data te herschrijven:

-- Kolom toevoegen (werkt direct, geen rewrite)
ALTER TABLE orders ADD COLUMN discount DECIMAL(5,2);

-- Kolom hernoemen
ALTER TABLE orders RENAME COLUMN amount TO order_amount;

-- Kolom verplaatsen
ALTER TABLE orders ALTER COLUMN discount AFTER order_amount;

-- Kolom verwijderen (data blijft, maar wordt niet meer gelezen)
ALTER TABLE orders DROP COLUMN legacy_flag;

Row-level Deletes & Updates (Copy-on-Write vs Merge-on-Read)

Iceberg ondersteunt twee strategieën voor updates en deletes:

-- Copy-on-Write (CoW): herschrijft bestanden bij elke update
-- → Sneller lezen, langzamer schrijven
-- Goed voor: batch updates, rapportage-tables

-- Merge-on-Read (MoR): slaat delete/update markers op
-- → Sneller schrijven, lezen vereist merge
-- Goed voor: frequente CDC / streaming updates

-- MERGE INTO (upsert) — werkt in Spark, Trino, Flink
MERGE INTO orders t
USING updates s ON t.order_id = s.order_id
WHEN MATCHED AND s.status = 'cancelled' THEN DELETE
WHEN MATCHED THEN UPDATE SET t.amount = s.amount, t.status = s.status
WHEN NOT MATCHED THEN INSERT *;

Lokaal testen met Spark + Iceberg

# pip install pyspark
# Download Iceberg Spark runtime JAR

from pyspark.sql import SparkSession

spark = SparkSession.builder \
    .appName("IcebergDemo") \
    .config("spark.jars.packages",
            "org.apache.iceberg:iceberg-spark-runtime-3.5_2.12:1.5.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()

# Tabel aanmaken
spark.sql("""
  CREATE TABLE local.db.orders (
    order_id BIGINT,
    customer_id BIGINT,
    amount DECIMAL(10,2),
    order_date DATE
  )
  USING iceberg
  PARTITIONED BY (months(order_date))
""")

# Data invoegen
spark.sql("""
  INSERT INTO local.db.orders VALUES
    (1, 101, 99.95, DATE '2026-03-01'),
    (2, 102, 249.00, DATE '2026-03-15'),
    (3, 101, 59.99, DATE '2026-03-20')
""")

# Query met time travel
spark.sql("SELECT * FROM local.db.orders.snapshots").show()
spark.sql("SELECT * FROM local.db.orders").show()

Iceberg op AWS: Glue + Athena

-- In AWS Glue Data Catalog / Athena:

-- Tabel aanmaken als Iceberg
CREATE TABLE orders (
  order_id BIGINT,
  customer_id BIGINT,
  amount DOUBLE,
  order_date DATE
)
LOCATION 's3://mijn-datalake/orders/'
TBLPROPERTIES (
  'table_type' = 'ICEBERG',
  'format' = 'parquet',
  'write_compression' = 'snappy'
);

-- Data invoegen (geen S3 copy nodig!)
INSERT INTO orders VALUES (1, 101, 99.95, DATE '2026-03-21');

-- UPDATE (Athena engine v3 + Iceberg)
UPDATE orders
SET amount = 89.95
WHERE order_id = 1;

-- DELETE
DELETE FROM orders WHERE order_id = 1;

-- Time travel
SELECT * FROM orders
FOR SYSTEM_TIME AS OF TIMESTAMP '2026-03-20 12:00:00';

Migratie van Parquet naar Iceberg (in-place)

-- Bestaande Parquet tabel migreren naar Iceberg
-- ZONDER data te kopiëren (alleen metadata aanmaken)

-- In Spark:
from pyspark.sql import SparkSession

# Migreer bestaande Parquet tabel
spark.sql("""
  CALL catalog.system.migrate(
    'db.bestaande_parquet_tabel'
  )
""")

-- Of via snapshot (veiliger: origineel blijft intact)
spark.sql("""
  CALL catalog.system.snapshot(
    source_table => 'parquet.db.orders',
    table => 'iceberg.db.orders'
  )
""")

-- Valideer de migratie
spark.sql("SELECT COUNT(*) FROM iceberg.db.orders").show()
spark.sql("SELECT * FROM iceberg.db.orders.snapshots").show()

Tabel Onderhoud: Compaction & Cleanup

-- Kleine bestanden samenvoegen (compaction)
CALL catalog.system.rewrite_data_files(
  table => 'db.orders',
  strategy => 'sort',
  sort_order => 'order_date ASC'
);

-- Verouderde snapshots verwijderen (data cleanup)
CALL catalog.system.expire_snapshots(
  table => 'db.orders',
  older_than => TIMESTAMP '2026-03-01 00:00:00',
  retain_last => 5
);

-- Verwijder ongebruikte metadata en orphan files
CALL catalog.system.remove_orphan_files(
  table => 'db.orders',
  older_than => TIMESTAMP '2026-03-01 00:00:00'
);

-- Bekijk tabel statistieken
SELECT * FROM db.orders.files LIMIT 20;
SELECT * FROM db.orders.partitions;