Case 1 — Lakehouse Migration: Partitioning & Querying with Apache Iceberg

# Case 1 — Lakehouse Migration: Partitioning & Querying with Apache Iceberg ## Executive Context & Business Prompt You are a Senior Data Engineer at **NexTrack Logistics**, a global freight forwarding platform. The company ingests over 150 million GPS telemetry and shipment tracking records per day directly from IoT sensors into Amazon S3 as raw Parquet files. The current raw S3 data lake suffers from severe production failure modes: 1. **Concurrent Write Corruption:** Nightly append batch runs conflict with near-real-time streaming ingesters, causing Trino and Athena analytical queries to fail with dirty reads and partial partition errors. 2. **Directory Partition Scan Overhead:** Querying by date and hour requires scanning thousands of tiny subdirectories, ballooning Athena costs to $4,500/month. 3. **Accidental Deletes with No Recovery:** A junior engineer recently issued an unpartitioned cleanup script that wiped 3 days of historical tracking data, requiring a 14-hour tape restore. VP of Engineering Prompt: > *"We need ACID transactions on our S3 data lake without migrating data into a costly proprietary warehouse. Migrate our telemetry fact dataset into an open table format using Apache Iceberg. We need hidden partitioning by hour, proven snapshot time-travel recovery, and an automated file compaction routine."* --- ## Explicit Portfolio Deliverables 1. **Iceberg Table DDL Script (`iceberg_telemetry_ddl.py` or `.sql`):** - Create table `lakehouse.logistics.fact_shipment_telemetry` backed by Apache Iceberg over S3. - Configure hidden partitioning on `event_timestamp` by day and hour without synthetic date columns. - Set table properties for Parquet compression (`zstd`) and target file size (`134217728` bytes / 128MB). 2. **Time-Travel Disaster Recovery Script (`iceberg_time_travel.py`):** - Demonstrate inserting a test batch of shipment tracking records. - Simulate an accidental batch deletion (`DELETE FROM fact_shipment_telemetry WHERE event_date = '2026-09-15'`). - Query the table snapshot prior to deletion using `FOR SYSTEM_VERSION AS OF` or `FOR SYSTEM_TIME AS OF` to prove zero data loss. - Revert the table to the healthy snapshot. 3. **Automated Maintenance & Compaction Routine (`iceberg_compaction.py`):** - Execute an in-place bin-pack rewrite (`CALL system.rewrite_data_files()`) to merge streaming small files into target 128MB files. - Run snapshot expiration (`CALL system.expire_snapshots()`) retaining 7 days of audit history while deleting orphaned metadata. --- ## Starter Code Scaffold ```python from pyspark.sql import SparkSession from datetime import datetime # Initialize Spark Session configured with Apache Iceberg catalog spark = SparkSession.builder \ .appName("IcebergLakehouseMigration") \ .config("spark.sql.extensions", "org.apache.iceberg.spark.extensions.IcebergSparkSessionExtensions") \ .config("spark.sql.catalog.lakehouse", "org.apache.iceberg.spark.SparkCatalog") \ .config("spark.sql.catalog.lakehouse.type", "hadoop") \ .config("spark.sql.catalog.lakehouse.warehouse", "s3a://nextrack-data-lakehouse/warehouse") \ .getOrCreate() # 1. TODO: Author DDL with hidden partitioning spark.sql(""" CREATE TABLE IF NOT EXISTS lakehouse.logistics.fact_shipment_telemetry ( tracking_id STRING, shipment_id STRING, carrier_code STRING, latitude DOUBLE, longitude DOUBLE, speed_kmh FLOAT, event_timestamp TIMESTAMP ) USING iceberg PARTITIONED BY (days(event_timestamp), carrier_code) TBLPROPERTIES ( 'write.format.default' = 'parquet', 'write.parquet.compression-codec' = 'zstd', 'write.target-file-size-bytes' = '134217728' ) """) # 2. TODO: Implement Time-Travel verification script # 3. TODO: Implement Maintenance rewrite_data_files routine ``` --- ## 100-Point Grading Rubric | Criteria | Points | Passing Standard | |---|:---:|---| | **1. Iceberg DDL & Partitioning Spec** | 25 pts | Correctly implements hidden partitioning using Iceberg partition transforms (`days()`, `hours()`) without redundant string columns. Sets target file size and zstd compression. | | **2. ACID Concurrent Write Guarantees** | 25 pts | Verifies snapshot isolation during concurrent simulated write workloads. Eliminates partial read anomalies. | | **3. Time-Travel & Snapshot Recovery** | 25 pts | Successfully executes point-in-time time-travel queries against snapshot IDs or timestamps; restores historical records after simulated disaster. | | **4. Maintenance & Compaction Optimization** | 25 pts | Implements automated `rewrite_data_files` compaction to eliminate the small-file problem and configures `expire_snapshots` for storage cost hygiene. |