Data Data Lake

Data Lake

Purpose

Design and operate ACID-compliant data lakes on object storage using Delta Lake, Apache Iceberg, or Apache Hudi. Manage table formats, optimize storage layout, enforce schema evolution, and implement time-travel and CDC patterns.

Agent Protocol

Trigger

Exact user phrases: "Delta Lake", "Apache Iceberg", "Apache Hudi", "Lakehouse", "ACID on lake", "time travel", "table format", "compaction", "Z-order", "Hilbert curve", "vacuum", "CDC", "merge-on-read", "copy-on-write", "manifest file", "metadata layer", "schema evolution", "optimize table", "incremental query".

Input Context

Before activating, verify:

• Object storage backend (S3, ADLS, GCS, MinIO)
• Table format preference (Delta, Iceberg, Hudi)
• Compute engine (Spark, Flink, Trino, Presto, Hive)
• Write pattern (append-heavy, update-heavy, CDC stream)
• Query pattern (OLAP, incremental, point lookup, full scan)
• Partition strategy (date, categorical, Z-order dimensions)

Output Artifact

Lake architecture with table format selection, compaction strategy, optimization plan.

Response Format

Table Format: {Delta Lake | Apache Iceberg | Apache Hudi}
Storage: {S3 | ADLS | GCS | MinIO}
Write Mode: {copy-on-write | merge-on-read}
Partition: {column, granularity}
Optimize: {Z-order by columns, compaction interval}
Vacuum: {retention hours}

-- DDL / optimization commands

No preamble. No postamble. No explanations. No filler/hedging/transitions. Compress output — why use many token when few do trick.

Completion Criteria

• Table format selected with trade-off analysis
• ACID guarantees documented (concurrent readers/writers)
• Write mode (CoW vs MoR) selected per table
• Partition strategy defined
• Compaction policy set (file size target, interval)
• Vacuum retention period configured
• Schema evolution rules defined

Max Response Length

250 lines of config.

Workflow

Step 1: Select Table Format

Format Comparison Matrix

Format Selection Decision Tree

Primary query engine?
├── Spark (only), Databricks ecosystem → Delta Lake
├── Multi-engine (Spark + Trino + Flink)
│   ├── Partition evolution needed → Apache Iceberg
│   └── No partition evolution → Delta Lake or Iceberg
├── Heavy CDC / update workloads
│   ├── Need native CDC → Apache Hudi (MOR)
│   └── General CDC → Apache Iceberg with merge
└── Hive/Presto/Trino only
    ├── Iceberg (best Hive/Trino support)
    └── Hudi (older Hive support)

Write pattern?
├── Append-heavy (logs, events) → Any format
├── Update-heavy (CDC, GDPR deletions) → Hudi (MOR) or Iceberg
└── Deletes (GDPR right to erasure) → Iceberg or Delta

Step 2: Configure ACID Guarantees

Delta Lake Concurrency

# Delta Lake ACID config
delta.autoOptimize.autoCompact: true
delta.autoOptimize.optimizeWrite: true
delta.maxRetryCommitAttempts: 10000000  # Retry on conflict
delta.tuneFileSizesForRewrites: true

# Concurrent write conflict handling
# Delta uses optimistic concurrency — retries on conflict
# For high-contention tables, use partition-level writes
# Avoid writing to same partition from multiple jobs simultaneously

# Isolation level
delta.isolationLevel: Serializable  # Default (most strict)
# delta.isolationLevel: WriteSerializable  # Less strict, better concurrency

Iceberg Concurrency

# Iceberg catalog-level conflict handling
# Iceberg uses optimistic concurrency at catalog level
# Retry mechanism built into Spark/Flink Iceberg integrations
# For high-contention scenarios, use:
  1. Partition-level writes
  2. Retry with exponential backoff (built-in)
  3. Serializable isolation (default)

# Iceberg catalog properties
iceberg:
  catalog:
    type: hive  # or nessie, glue, jdbc, rest
    lock-impl: in-memory  # For testing only
    # Production: use catalog-native locking (HMS Lock, DynamoDB)
    warehouse: s3://data-lake/warehouse

Step 3: Write Mode Selection

Copy-on-Write (CoW)

Writes rewrite entire Parquet files. Best for: read-heavy workloads, append-only tables, smaller tables, OLAP queries. Pros: no merge at read time, simple file layout, compact storage. Cons: write amplification (rewrite whole file on single row update).

Merge-on-Read (MoR)

Writes append delta logs (Avro/Parquet), compacted later. Best for: update-heavy workloads, CDC ingestion, large tables with frequent updates, tables where write speed matters more than read speed. Pros: fast writes, no write amplification. Cons: read overhead merging base + delta files.

# Hudi MOR configuration
hoodie.datasource.write.table.type: MERGE_ON_READ
hoodie.table.type: MERGE_ON_READ
hoodie.compact.inline: true  # Inline compaction
hoodie.compact.inline.max.delta.commits: 5  # Compact every 5 commits

# Iceberg: use merge-in-read (not separate write mode)
# Iceberg handles all updates via merge-on-read automatically

Step 4: Partition Strategy

Partition Design Rules

Partition by: low-to-medium cardinality columns (< 1000 partitions for daily, < 365 for daily year). Prefer: date-based partitioning (event_date, year/month/day) for time-series data. Avoid: high-cardinality columns as partition keys (customer_id, order_id). Iceberg hidden partitioning: define partition transforms and Iceberg manages partition layout transparently.

-- Delta Lake
CREATE TABLE events (
    event_id STRING,
    event_type STRING,
    event_date DATE,
    payload STRUCT<
        user_id: STRING,
        page: STRING,
        duration: INT
    >
)
USING DELTA
PARTITIONED BY (event_date)
LOCATION 's3://data-lake/events/';

-- Apache Iceberg with hidden partitioning
CREATE TABLE events (
    event_id STRING,
    event_type STRING,
    event_date DATE,
    payload STRUCT<user_id: STRING, page: STRING, duration: INT>
)
USING ICEBERG
PARTITIONED BY (days(event_date))
LOCATION 's3://data-lake/events/';

-- Hudi
CREATE TABLE events (
    event_id STRING,
    event_type STRING,
    event_date DATE,
    payload STRING  -- JSON
)
USING HUDI
PARTITIONED BY (event_date)
OPTIONS (
    primaryKey = 'event_id',
    preCombineField = 'event_date',
    hoodie.datasource.write.table.type = 'COPY_ON_WRITE'
)
LOCATION 's3://data-lake/events/';

Step 5: File Layout Optimization

Target File Size

# Target: 128MB - 1GB files (balance parallelism with overhead)
# For Delta:
delta.targetFileSize: 256mb
delta.tuneFileSizesForRewrites: true

# For Iceberg:
write.target-file-size-bytes: 268435456  # 256MB

# For Hudi:
hoodie.parquet.max.file.size: 268435456
hoodie.parquet.small.file.limit: 134217728  # 128MB - files below this merged

Compaction

-- Delta Lake compact
OPTIMIZE events ZORDER BY (event_type, user_id);

-- Iceberg compact
CALL catalog.system.rewrite_data_files(
    table => 'events',
    strategy => 'binpack',
    options => map('target-file-size-bytes', '268435456')
);

-- Iceberg expire snapshots
CALL catalog.system.expire_snapshots('events', TIMESTAMP '2026-01-01 00:00:00');

-- Hudi compaction (inline)
-- Configured at write time:
hoodie.compact.inline: true
hoodie.compact.inline.max.delta.commits: 5
hoodie.compact.inline.max.delta.seconds: 3600

Z-Order / Hilbert Clustering

-- Delta Z-order: reduces data scanned for queries filtering on clustered columns
OPTIMIZE events ZORDER BY (event_type, user_id);
-- Z-order effect: queries on event_type or user_id scan fewer files
-- Best for: 2-4 columns frequently used in WHERE clauses together

-- Iceberg sort order (v2+)
CREATE TABLE events (
    ...
) USING ICEBERG
PARTITIONED BY (days(event_date))
ORDER BY (event_type, user_id);

-- Hudi clustering
CALL run_clustering(
    table => 'events',
    order => 'event_type,user_id',
    target_file_max_size => 268435456
);

Step 6: Vacuum / Snapshot Expiration

# Delta VACUUM default retention: 7 days
# Keep enough for concurrent read isolation + time travel queries
vacuum:
  retention_hours: 168  # 7 days
  # For compliance: extend based on audit requirements
  # For testing environments: reduce to 24h
  # NEVER vacuum concurrently with active write operations

# Iceberg snapshot expiration
expire_snapshots:
  retain_last_n: 10     # Keep 10 most recent
  older_than: 7 days    # Delete snapshots older than 7 days
  # Run daily as maintenance job
  # Spark SQL: CALL catalog.system.expire_snapshots('table', older_than => 7)

# Hudi cleaner
hoodie.cleaner.policy: KEEP_LATEST_COMMITS
hoodie.cleaner.commits.retained: 10  # Keep last 10 commits

Step 7: Schema Evolution

Evolution Allowed Operations

-- Delta Lake schema evolution
ALTER TABLE events ADD COLUMN new_field STRING;
ALTER TABLE events ALTER COLUMN old_field DROP;
ALTER TABLE events RENAME COLUMN old_field TO new_field;

-- Iceberg schema evolution
ALTER TABLE events ADD COLUMN new_field STRING;
ALTER TABLE events DROP COLUMN old_field;
ALTER TABLE events RENAME COLUMN old_field TO new_field;
-- Iceberg reorders columns:
ALTER TABLE events ALTER COLUMN col2 FIRST;
ALTER TABLE events ALTER COLUMN col3 AFTER col1;

-- Hudi schema evolution (Spark SQL)
ALTER TABLE events ADD COLUMNS (new_field STRING);

Step 8: Time Travel

-- Delta Lake time travel
SELECT * FROM events VERSION AS OF 12345;        -- By version number
SELECT * FROM events TIMESTAMP AS OF '2026-05-01'; -- By timestamp

-- Restore to previous version
RESTORE TABLE events TO VERSION AS OF 12345;

-- Iceberg time travel
SELECT * FROM events FOR SYSTEM_TIME AS OF '2026-05-01 12:00:00';  -- By timestamp
SELECT * FROM events FOR SYSTEM_VERSION AS OF 12345;                -- By snapshot ID

-- Hudi time travel
SELECT * FROM events TIMESTAMP AS OF '2026-05-01 12:00:00';

Step 9: Incremental Queries

-- Delta incremental query (Spark Streaming source)
spark.readStream.format("delta")
    .option("startingVersion", "100")
    .table("events")

-- Delta change data feed
ALTER TABLE events SET TBLPROPERTIES (delta.enableChangeDataFeed = true);
SELECT * FROM table_changes('events', 12345, 12350);

-- Iceberg incremental read
SELECT * FROM events.incremental(100)  -- Spark incremental query

-- Hudi incremental pull
SELECT * FROM events WHERE _hoodie_commit_time > '20260101000000';

Step 10: Performance Optimization

Read Optimization

# Parallelism and file scanning
spark.sql.files.maxPartitionBytes: 256MB     # Max bytes per partition
spark.sql.files.openCostInBytes: 4MB          # Cost of opening a file
spark.sql.files.minPartitionNum: 1            # Min partitions

# Iceberg read optimizations
read.split.target-size: 268435456             # 256MB split target
read.split.planning-lookback: 10              # Planned splits count
read.split.open-file-cost: 4194304            # 4MB open cost

Write Optimization

# Parallel writes
spark.sql.shuffle.partitions: 400            # Adjust for data volume
spark.sql.hive.convertMetastoreParquet: true

# Iceberg write optimizations
write.distribution-mode: hash                 # hash, range, none
write.wap.enabled: false                      # Write-Audit-Publish
write.merge.mode: merge-on-read               # Default for Iceberg updates

# Delta write optimizations
delta.autoOptimize.autoCompact: true
delta.autoOptimize.optimizeWrite: true

Rules

• Always use open table formats — never raw Parquet/ORC on object stores
• Partition by date for time-series data — enable partition pruning
• Compaction target: 128MB-1GB files for optimal read parallelism
• Vacuum retention: 7 days minimum for time travel and concurrent reads
• Enable Z-order/Hilbert clustering for multi-column filter patterns
• Use hidden partitioning (Iceberg) for transparent partition management
• Monitor small files — run compaction when avg file size < 64MB
• Run vacuum/snapshot expiration as scheduled maintenance jobs
• Enable change data feed for incremental downstream consumers
• Test schema changes on staging before production rollout
• Use column-level statistics for CBO on query engines

Lake Maintenance Jobs

# Recommended maintenance schedule for lake tables
maintenance_jobs:
  - name: compact_small_files
    schedule: "0 2 * * *"  # Daily at 2 AM
    tables:
      - pattern: "prod.*"  # All production tables
      - except: ["prod.ref.*"]  # Skip reference tables
    target_file_size: 256MB
  
  - name: expire_snapshots
    schedule: "0 3 * * *"  # Daily at 3 AM
    retention: 7 days
    tables:
      - pattern: "*"
  
  - name: remove_orphan_files
    schedule: "0 4 * * 0"  # Weekly on Sunday
    tables:
      - pattern: "*"
  
  - name: update_statistics
    schedule: "0 5 * * *"  # Daily at 5 AM
    tables:
      - pattern: "prod.*"
    compute_stats: true
    compute_column_stats: true

Rules

• Always use open table formats — never raw Parquet/ORC on object stores
• Partition by date for time-series data — enable partition pruning
• Compaction target: 128MB-1GB files for optimal read parallelism
• Vacuum retention: 7 days minimum for time travel and concurrent reads
• Enable Z-order/Hilbert clustering for multi-column filter patterns
• Use hidden partitioning (Iceberg) for transparent partition management
• Monitor small files — run compaction when avg file size < 64MB
• Run vacuum/snapshot expiration as scheduled maintenance jobs
• Enable change data feed for incremental downstream consumers
• Test schema changes on staging before production rollout
• Use column-level statistics for CBO on query engines
• Never run vacuum concurrently with active write operations
• Document table format decision in ADR with specific rationale
• Monitor write amplification for update-heavy tables (prefer MoR)
• Use version catalog (Nessie) for Git-like branching on the lake

References

Coming soon.

Architecture Decision Trees

Lake Table Format Selection
├── ACID transactions on lake?
│   ├── Yes → Delta Lake / Iceberg / Hudi
│   └── No → Hive-style partitions (legacy)
├── Cross-engine compatibility?
│   ├── Yes → Iceberg (Spark, Flink, Trino, Athena)
│   ├── Databricks ecosystem? → Delta Lake
│   └── No → Hudi (mature upsert/deletes)
├── Real-time ingestion required?
│   ├── Yes → Hudi (Copy-on-Write + Merge-on-Read)
│   └── No → Iceberg (batch optimized)
└── Schema evolution needed?
    ├── Yes → Iceberg (add/drop/rename columns freely)
    └── No → Delta Lake

Decision criteria: Evaluate engine ecosystem, transaction requirements, latency SLA, and schema flexibility.

Implementation Patterns

Iceberg Table Maintenance

# data_lake/iceberg_maintenance.py
from pyspark.sql import SparkSession

class IcebergMaintenance:
    def __init__(self, spark: SparkSession, table: str):
        self.spark = spark
        self.table = table

    def expire_snapshots(self, retain_days: int = 7):
        self.spark.sql(f"""
            CALL {self.table}.expire_snapshots(
                older_than => TIMESTAMP '{datetime.now() - timedelta(days=retain_days)}',
                retain_last => 5
            )
        """)

    def compact_data(self, target_size_mb: int = 256):
        self.spark.sql(f"""
            CALL {self.table}.rewrite_data_files(
                target_file_size_bytes => {target_size_mb * 1024 * 1024}
            )
        """)

    def rewrite_manifests(self):
        self.spark.sql(f"CALL {self.table}.rewrite_manifests()")

Lake Catalog Registration

# data_lake/catalog_registration.yml
catalog:
  type: rest
  uri: http://nessie:19120/api/v1
  warehouse: s3://data-lake/warehouse
  default_branch: main
  tables:
    - name: bronze.orders
      format: iceberg
      partitioning:
        - identity: ingestion_date
        - bucket(16): order_id
    - name: silver.customers
      format: delta
      partitioning:
        - identity: region

Production Considerations

• Compaction scheduling: Schedule data compaction during low-traffic windows; monitor file count per partition.
• Vacuum retention: Keep snapshot history for 7 days for time-travel queries; vacuum older snapshots.
• Write amplification: Prefer Copy-on-Write for read-heavy workloads; Merge-on-Read for write-heavy.
• Partition evolution: Iceberg supports hidden partitioning; avoid repartitioning large tables.
• Catalog HA: Deploy Nessie or AWS Glue Catalog with high availability; cache metadata locally.
• Storage tiering: Use S3 Intelligent-Tiering for cost optimization; expire old data to Glacier.

Anti-Patterns

Performance Optimization

• Partition pruning: Align partitions with query filter patterns (date, region); use Iceberg hidden partitions.
• File clustering: Run Z-order clustering on frequently filtered non-partition columns (Iceberg sort order).
• Metadata caching: Cache Iceberg/Delta manifests in memory (local SSD) to reduce S3 metadata requests.
• Committed write ordering: Use sorted writes to align with partition boundaries for faster merges.
• Parallel manifest reads: Increase metadata parallelism via spark.sql.iceberg.worker.numThreads.

Security Considerations

• Table ACLs: Use Ranger/Atlas for lake table-level authorization; restrict DROP TABLE permissions.
• Data encryption: Enable S3 server-side encryption (SSE-S3 or SSE-KMS) for lake storage.
• Audit trail: Log all DDL operations, compaction runs, and schema changes to AWS CloudTrail.
• Credential management: Use IAM roles for Spark/Trino access to lake storage; rotate keys.
• Network isolation: Deploy lake storage in VPC endpoints; no public S3 access to data lake buckets.

Handoff

data-data-platform for overall lake architecture data-data-warehouse for warehouse layer on top of lake data-distributed-storage for storage infrastructure details