S052 · Lakehouse — Delta / Iceberg / Hudi, ACID on Files

"""
lakehouse_demo.py — Create a Delta table, evolve it, time-travel through history.
 
Requires: pip install "deltalake>=0.17" pandas pyarrow
Runs entirely on your laptop — no Spark, no S3.
"""
from __future__ import annotations
 
import json
import shutil
from pathlib import Path
 
import pandas as pd
from deltalake import DeltaTable, write_deltalake
 
TABLE = Path("/tmp/lakehouse_demo/orders")
 
 
def reset() -> None:
    """Start from a clean slate so the demo is idempotent."""
    if TABLE.exists():
        shutil.rmtree(TABLE)
    TABLE.parent.mkdir(parents=True, exist_ok=True)
 
 
def commit_0_initial_load() -> None:
    """Commit 0 — three orders on Monday."""
    df = pd.DataFrame(
        [
            {"order_id": 1, "customer": "alice", "amount": 42.00, "day": "mon"},
            {"order_id": 2, "customer": "bob",   "amount": 19.50, "day": "mon"},
            {"order_id": 3, "customer": "carol", "amount": 88.00, "day": "mon"},
        ]
    )
    write_deltalake(str(TABLE), df, mode="overwrite", partition_by=["day"])
    print(f"[commit 0] wrote {len(df)} rows")
 
 
def commit_1_append_tuesday() -> None:
    """Commit 1 — two more orders on Tuesday."""
    df = pd.DataFrame(
        [
            {"order_id": 4, "customer": "dave", "amount": 12.00, "day": "tue"},
            {"order_id": 5, "customer": "eve",  "amount": 66.75, "day": "tue"},
        ]
    )
    write_deltalake(str(TABLE), df, mode="append", partition_by=["day"])
    print(f"[commit 1] appended {len(df)} rows")
 
 
def commit_2_delete_customer(target: str) -> None:
    """Commit 2 — GDPR-style delete: remove everything for `target`."""
    dt = DeltaTable(str(TABLE))
    deleted = dt.delete(predicate=f"customer = '{target}'")
    # delete() returns operation metrics as a dict
    print(f"[commit 2] deleted {deleted['num_deleted_rows']} rows for {target}")
 
 
def show_history() -> None:
    dt = DeltaTable(str(TABLE))
    print("\n=== history ===")
    for entry in dt.history():
        print(
            f"v{entry['version']:>2}  {entry['operation']:<10}"
            f"  ts={entry.get('timestamp','?')}"
        )
 
 
def time_travel(version: int) -> None:
    dt = DeltaTable(str(TABLE), version=version)
    df = dt.to_pandas()
    print(f"\n=== SELECT * VERSION AS OF {version} ({len(df)} rows) ===")
    print(df.to_string(index=False))
 
 
def peek_transaction_log() -> None:
    """Read the raw JSON commits — the whole point of ‘it's just files’."""
    log_dir = TABLE / "_delta_log"
    commits = sorted(log_dir.glob("*.json"))
    print(f"\n=== _delta_log/ ({len(commits)} commit files) ===")
    for path in commits:
        actions = [json.loads(line) for line in path.read_text().splitlines()]
        kinds = ", ".join(sorted({next(iter(a)) for a in actions}))
        print(f"  {path.name}: {len(actions)} actions [{kinds}]")
 
 
def main() -> None:
    reset()
    commit_0_initial_load()
    commit_1_append_tuesday()
    commit_2_delete_customer("bob")
 
    show_history()
    time_travel(0)  # Monday-only, 3 rows
    time_travel(1)  # Monday+Tuesday, 5 rows
    time_travel(2)  # bob is gone, 4 rows
    peek_transaction_log()
 
 
if __name__ == "__main__":
    main()

pip install "deltalake>=0.17" pandas pyarrow
python lakehouse_demo.py

import polars as pl
lf = pl.scan_delta("/tmp/lakehouse_demo/orders")
print(lf.filter(pl.col("amount") > 20).collect())