Lakehouse Architecture

┌─────────────────────────────────────────────────────────────────┐
│                    QUERY / COMPUTE LAYER                        │
│  Spark (batch + streaming)   Databricks SQL Warehouse           │
│  Trino / Athena (ad hoc)     dbt (transformations)              │
│  Flink (streaming)           TensorFlow / PyTorch (ML)          │
│  All engines speak the SAME table format protocol               │
└──────────────────────────┬──────────────────────────────────────┘
                           │ reads/writes through table format API
┌──────────────────────────▼──────────────────────────────────────┐
│                    OPEN TABLE FORMAT LAYER                      │
│  Delta Lake (Databricks)  │  Apache Iceberg  │  Apache Hudi     │
│  Provides: ACID transactions, time travel, schema enforcement,  │
│  row-level DELETE/UPDATE, partition evolution, data skipping    │
│  Implemented as: transaction log + Parquet data files           │
└──────────────────────────┬──────────────────────────────────────┘
                           │ raw files on object storage
┌──────────────────────────▼──────────────────────────────────────┐
│                    OBJECT STORAGE LAYER                         │
│  AWS S3  │  Azure ADLS Gen2  │  Google Cloud Storage            │
│  Petabyte-scale, cheap ($23/TB/month), 11 nines durability,     │
│  all data stored as open-format Parquet — no vendor lock-in     │
└─────────────────────────────────────────────────────────────────┘

┌─────────────────────────────────────────────────────────────────┐
│                    GOVERNANCE LAYER (cross-cutting)             │
│  Unity Catalog (Databricks)  │  Apache Polaris (open source)    │
│  AWS Glue Data Catalog       │  Nessie (open source)            │
│  Table discovery, column-level access control, cross-engine     │
│  lineage, audit logging                                         │
└─────────────────────────────────────────────────────────────────┘

KEY INSIGHT: every layer is replaceable.
Replace Spark with Trino: same data, same governance, same ACID.
Replace Delta Lake with Iceberg: change metadata format, same Parquet data.
Replace S3 with ADLS: Iceberg adapts its path format.
This composability is what the lakehouse pattern uniquely provides.

The transaction log (_delta_log/) is the source of truth for table state.
Parquet data files are just bytes — they have no meaning without the log.
The log determines which files are part of the current table version.

ATOMICITY (all or nothing):
  Writer's steps:
    1. Write new Parquet files to the table directory (side effect-free)
       These files EXIST on S3 but are INVISIBLE to readers — not in the log yet
    2. Write a new commit entry to _delta_log/000...042.json
       This is a SINGLE S3 PUT — an atomic operation
  If step 1 fails: no log entry written → files invisible → table unchanged
  If step 2 fails: log entry not written → files invisible → table unchanged
  If step 2 succeeds: both files visible simultaneously → atomic commit ✓

CONSISTENCY (schema enforced at every commit):
  Before writing the log entry, Delta checks the new data's schema is
  compatible with the table schema, types match (or evolution is explicitly
  allowed), and required columns are not missing.
  If the check fails: log entry rejected → no data written → consistent ✓

ISOLATION (concurrent writers do not corrupt each other):
  Delta uses optimistic concurrency control:
    Writer A reads current table version: v41
    Writer B reads current table version: v41
    Writer A writes Parquet files, attempts to commit log entry v42
    Writer B writes Parquet files, attempts to commit log entry v42
    → S3 atomic PUT: only one can succeed (Delta uses conditional PUT or
      atomic rename strategies, depending on the storage backend)
    Writer A succeeds → table is now v42
    Writer B detects conflict (log entry v42 already exists):
      → If appending non-overlapping partitions: REBASE and commit as v43
      → If touching overlapping partitions: ABORT and retry
  Result: only valid committed states are visible → isolation ✓

DURABILITY (committed data survives failures):
  S3 has 99.999999999% (11 nines) durability for stored objects. Once the
  log entry is committed, the data is durable. Log entries are immutable —
  once written, never modified. Recovery after failure: read the log from
  the beginning (or last checkpoint) to reconstruct the current table state ✓

Instead of a sequential JSON log, Iceberg uses a tree of metadata files:
  metadata/
    v1.metadata.json             ← snapshot list, schema history, partition spec
    snap-001-manifest-list.avro  ← list of manifest files for this snapshot
  manifests/
    manifest-001.avro            ← list of data files and their statistics
  data/
    part-00001.parquet, ...

COMMIT PROTOCOL:
  1. Write new data files (Parquet) — invisible until committed
  2. Write new manifest file listing new data files
  3. Write new manifest list referencing new manifest
  4. Atomically swap the metadata file pointer: v1.metadata.json → v2.metadata.json
     (catalog-level atomic pointer swap — varies by catalog implementation)
  5. If pointer swap succeeds: new snapshot is current state ✓

KEY DIFFERENCE FROM DELTA: Iceberg's tree of metadata objects (not a
sequential log) enables better performance for very large tables (millions
of files) — reading the manifest list is O(1) rather than scanning the full log.

UPDATE silver.orders SET status = 'delivered'
WHERE order_id = 9284751 AND order_date = '2026-03-17';

Step 1: Read micro-partition containing order_id 9284751
        (file: date=2026-03-17/part-00042.parquet)
Step 2: Apply update in memory: status changed to 'delivered'
Step 3: Write ENTIRE FILE with the update applied:
        new file: date=2026-03-17/part-00043.parquet (full partition)
Step 4: Commit new log entry:
        REMOVE: date=2026-03-17/part-00042.parquet
        ADD:    date=2026-03-17/part-00043.parquet

RESULT: the new snapshot shows part-00043 (updated), not part-00042 (old).
Old file stays on S3 until VACUUM removes it (supports time travel).

CoW write amplification: 1 row updated in a 128 MB file → 128 MB rewritten.
Expensive at write time for high-update-rate tables — but reads are fast
(no merge needed at read time).

Instead of rewriting the full file on every update:
  Write a small "delete file" recording which rows are deleted
  Write a small "position delete" or "equality delete" file
  New data written as new small files
On READ: the engine merges base files + delete files → current state

MoR write cost: cheap (write small delta files only)
MoR read cost:  more expensive (must merge delete files on every read)
Use MoR when: high write velocity, low read frequency (e.g. CDC ingestion)
Use CoW when: high read frequency, moderate write rate (e.g. analytics tables)

-- Snowflake / Databricks / BigQuery equivalent:
MERGE INTO silver.orders AS target
USING (
    SELECT order_id, status, amount, updated_at
    FROM bronze.orders_cdc WHERE _bronze_date = '2026-03-17'
) AS source
ON target.order_id = source.order_id
WHEN MATCHED AND target.updated_at < source.updated_at
  THEN UPDATE SET status = source.status, amount = source.amount,
                  updated_at = source.updated_at
WHEN NOT MATCHED
  THEN INSERT (order_id, status, amount, updated_at)
       VALUES (source.order_id, source.status, source.amount, source.updated_at);

Delta Lake executes this as:
1. Hash join target and source on order_id
2. For matched rows where condition is true: mark old file for removal,
   write updated rows to a new file (CoW)
3. For unmatched rows: write new rows to a new file
4. Commit: REMOVE old files, ADD new files in one atomic log entry

-- Query table at a specific version number:
SELECT * FROM silver.orders VERSION AS OF 41;

-- Query table at a specific timestamp:
SELECT * FROM silver.orders TIMESTAMP AS OF '2026-03-16 23:59:59';

-- Using the Spark API:
df = spark.read.format("delta").option("versionAsOf", 41) \
    .load("s3://freshcart-lake/silver/orders")
df = spark.read.format("delta").option("timestampAsOf", "2026-03-16 23:59:59") \
    .load("s3://freshcart-lake/silver/orders")

-- View table history:
DESCRIBE HISTORY silver.orders;
-- version 42: MERGE (2026-03-17 06:14:32) — 48,234 rows merged
-- version 41: MERGE (2026-03-16 06:11:47) — 47,892 rows merged
-- version  0: CREATE TABLE (2026-01-01 00:00:00)

-- Restore table to a previous version (non-destructive — creates a new commit):
RESTORE TABLE silver.orders TO VERSION AS OF 41;

-- Query at snapshot ID:
SELECT * FROM silver.orders FOR SYSTEM_VERSION AS OF 5765671814693002000;
-- Query at timestamp:
SELECT * FROM silver.orders FOR SYSTEM_TIME AS OF '2026-03-16 23:59:59';
-- View snapshots:
SELECT * FROM silver.orders.snapshots;
-- Rollback to snapshot:
CALL system.rollback_to_snapshot('freshcart.silver.orders', 5765671814693002000);

TIME TRAVEL RETENTION (Delta Lake):
  delta.logRetentionDuration (default 30 days) and
  delta.deletedFileRetentionDuration (default 7 days) control the window.
  VACUUM removes files older than the retention window.
  SET TBLPROPERTIES (delta.logRetentionDuration = 'interval 90 days')
  CAUTION: longer retention = more storage cost. Valuable for GDPR/audit,
  expensive for high-write tables.

1. DEBUG A PIPELINE BUG:
   A pipeline wrote wrong revenue figures on 2026-03-10, fixed on 2026-03-11.
   SELECT SUM(amount) FROM silver.orders TIMESTAMP AS OF '2026-03-10 23:59:59';
   -- compare to:
   SELECT SUM(amount) FROM silver.orders TIMESTAMP AS OF '2026-03-11 23:59:59';

2. REGULATORY AUDIT:
   Regulator asks for total active customer count as of Q4 end.
   SELECT COUNT(*) FROM silver.customers TIMESTAMP AS OF '2026-03-31 23:59:59';
   -- Returns the exact count from that date — provable, reproducible.

3. ML REPRODUCIBILITY:
   Reproduce the exact training dataset used on March 1 for an audit:
   df = spark.read.format("delta") \
       .option("timestampAsOf", "2026-03-01 00:00:00") \
       .load(silver_orders_path)

4. RECOVER FROM AN ACCIDENTAL DELETE:
   Someone ran DELETE FROM silver.orders WHERE store_id = 'ST001' by mistake.
   RESTORE TABLE silver.orders TO VERSION AS OF (current_version - 1);
   -- Table recovered to the state before the accidental delete.

s3://freshcart-lake/silver/orders/
├── metadata/
│   ├── v1.metadata.json, v2.metadata.json, v3.metadata.json (current)
│   └── snap-001/002/003-manifest-list.avro (per-snapshot manifest lists)
├── manifests/
│   └── manifest-001/002/003.avro (lists data files for each snapshot)
└── data/
    ├── date=2026-03-15/part-00001-abc123.parquet
    ├── date=2026-03-16/part-00001-def456.parquet
    └── date=2026-03-17/part-00001-ghi789.parquet

CATALOG: the external service that stores the current metadata pointer.
  Maps table_name → current metadata file path. Without the catalog, you
  cannot know which metadata file is current.
  Supported implementations: Hive Metastore, AWS Glue Data Catalog, Apache
  Nessie (git-like, with branching/tagging), REST Catalog (Tabular,
  Databricks Unity Catalog), JDBC Catalog (dev only).

1. CATALOG LOOKUP: client asks catalog for table "silver.orders"
   Catalog returns: metadata file = "v3.metadata.json"
2. READ METADATA FILE: contains current snapshot ID, schema history, spec
3. READ MANIFEST LIST: contains manifest files for this snapshot, each with
   partition range stats — apply partition pruning here
4. READ RELEVANT MANIFESTS: contains data files + per-file stats
   (min/max/null_count) — apply data file pruning
5. READ RELEVANT PARQUET FILES: predicate pushdown at the row group level

PERFORMANCE: steps 1-4 are metadata-only (small files, fast). Step 5 is
where actual data I/O happens. Well-pruned queries skip 90-99% of data
files → dramatic speedup.

PROBLEM IN DELTA LAKE:
  Table created with PARTITION BY (order_date) in 2024. By 2026: 50 TB,
  queries now filter by (store_id, order_date). Changing partition
  strategy in Delta requires a full table rewrite — 8-12 hours, hundreds
  of dollars, outage risk.

ICEBERG PARTITION EVOLUTION (no data rewrite needed):
  ALTER TABLE silver.orders DROP PARTITION FIELD months(order_date);
  ALTER TABLE silver.orders ADD PARTITION FIELD days(order_date);
  ALTER TABLE silver.orders ADD PARTITION FIELD identity(store_id);
  -- New data written with: partitioned by (day(order_date), store_id)

  Old data files still have month-based partition metadata in their
  manifests; new data files have (day, store_id) metadata. Each manifest
  records which spec was used for its files — a query prunes old manifests
  by month and new manifests by day+store, with no rewrite and no
  correctness gap.

Iceberg partition transforms partition by a derived value, not the raw
column: years(ts), months(ts), days(ts), hours(ts), bucket(N, col),
truncate(W, col).

These are HIDDEN from query writers — a query WHERE order_date =
'2026-03-17' is internally mapped by Iceberg to the days(2026-03-17)
partition. The user never needs to know the partition granularity.

BENEFIT: queries do not break when partition granularity changes. The same
query WHERE order_date BETWEEN '2026-03-01' AND '2026-03-17' scans all
March files under months(order_date) partitioning, but only March 1-17
files once re-partitioned to days(order_date) — same query, better
performance, no migration.

freshcart_prod.silver.orders               ← production Silver orders
freshcart_prod.gold.daily_revenue          ← production Gold metrics
freshcart_dev.silver.orders                ← dev environment (separate catalog)
freshcart_prod.ml_features.order_features  ← ML feature store tables

GRANT SELECT ON CATALOG freshcart_prod TO GROUP analysts;
GRANT SELECT ON SCHEMA  freshcart_prod.silver TO USER priya@freshcart.com;
GRANT SELECT ON TABLE   freshcart_prod.gold.daily_revenue TO GROUP finance;

-- Deny access to PII columns in Silver, grant only non-PII columns:
REVOKE SELECT ON TABLE freshcart_prod.silver.customers FROM GROUP analysts;
GRANT SELECT (customer_id, tier, city, lifetime_orders)
  ON TABLE freshcart_prod.silver.customers TO GROUP analysts;

-- COLUMN MASKING:
CREATE OR REPLACE MASKING POLICY mask_email AS (val STRING)
RETURNS STRING ->
  CASE WHEN is_account_group_member('data_engineers')
       THEN val                          -- engineers see raw email
       ELSE SHA2(val, 256)               -- others see hash
  END;

ALTER TABLE freshcart_prod.bronze.customers
ALTER COLUMN email SET MASKING POLICY mask_email;

-- ROW-LEVEL SECURITY:
CREATE OR REPLACE ROW ACCESS POLICY store_partition_policy
AS (store_id STRING) RETURNS BOOLEAN ->
    is_account_group_member('all_stores')
    OR store_id = current_user_store_id();

ALTER TABLE freshcart_prod.gold.store_performance
ADD ROW ACCESS POLICY store_partition_policy ON (store_id);
-- Store managers only see their own store's performance data

UNITY CATALOG LINEAGE (captured automatically from Spark queries,
Databricks SQL queries, and dbt runs via the Unity Catalog integration):
  SELECT * FROM system.information_schema.column_lineage
  WHERE target_table_name = 'daily_revenue';
  -- Returns: gold.daily_revenue.net_revenue ←
  --   silver.orders.order_amount, silver.orders.discount_amount

METASTORE ARCHITECTURE: one Databricks-managed metastore per cloud region
per account. Multiple workspaces share the same metastore → same
governance. Tables live on customer-owned S3/ADLS — Databricks never
holds the data itself, only the metastore endpoint.

ICEBERG REST CATALOG SPECIFICATION (standard REST API):
  GET  /v1/namespaces                              list all namespaces
  GET  /v1/namespaces/{ns}/tables                  list tables in namespace
  POST /v1/namespaces/{ns}/tables                  create table

  ANY engine implementing this REST client can use ANY catalog server —
  Spark, Flink, Trino, and Athena (via Glue Iceberg support) can all point
  at the same Polaris REST catalog. This is the open standard equivalent
  of Unity Catalog's proprietary API.

CONFIGURING SPARK TO USE POLARIS:
  spark.sql.catalog.freshcart = org.apache.iceberg.spark.SparkCatalog
  spark.sql.catalog.freshcart.type = rest
  spark.sql.catalog.freshcart.uri = https://polaris.freshcart.internal/api/catalog
  spark.sql.catalog.freshcart.warehouse = s3://freshcart-lake/iceberg

APACHE NESSIE — the git-like alternative: adds branching and tagging to
Iceberg catalogs. Create a branch for data experimentation without
touching production, then merge to main once validated. Used by Project
Nessie and Arctic (Dremio's managed Nessie offering).

from pyspark.sql import SparkSession
from pyspark.sql.functions import col, from_json, current_timestamp

spark = SparkSession.builder \
    .config("spark.sql.extensions", "io.delta.sql.DeltaSparkSessionExtension") \
    .getOrCreate()

orders_cdc = spark.readStream.format("kafka") \
    .option("kafka.bootstrap.servers", "kafka:9092") \
    .option("subscribe", "freshcart.cdc.public.orders") \
    .load()

def write_to_bronze(batch_df, batch_id):
    batch_df.select(from_json(col("value").cast("string"), order_schema).alias("e")) \
        .select("e.*").withColumn("_bronze_ts", current_timestamp()) \
        .withColumn("_batch_id", batch_id) \
        .write.format("delta").mode("append") \
        .option("mergeSchema", "true") \
        .save("s3://freshcart-lake/bronze/orders")

query = orders_cdc.writeStream.foreachBatch(write_to_bronze) \
    .option("checkpointLocation", "s3://freshcart-lake/checkpoints/orders_bronze") \
    .trigger(processingTime="1 minute").start()

# WHILE THIS STREAMING WRITE IS HAPPENING: dbt runs against silver.orders
# (reads Bronze) with NO conflict. Analysts query silver.orders with NO
# conflict. Delta Lake's optimistic concurrency handles concurrent readers.

changes = spark.readStream.format("delta") \
    .option("readChangeFeed", "true") \
    .option("startingVersion", "latest") \
    .load("s3://freshcart-lake/bronze/orders")
# changes includes: _change_type (insert/update_preimage/update_postimage/delete)

def merge_to_silver(batch_df, batch_id):
    from delta.tables import DeltaTable
    from pyspark.sql.window import Window
    from pyspark.sql.functions import row_number

    # Keep only the latest CDC event per order_id in this batch:
    latest = batch_df.filter(col("_change_type").isin("insert", "update_postimage")) \
        .withColumn("rn", row_number().over(
            Window.partitionBy("order_id").orderBy(col("updated_at").desc()))) \
        .filter(col("rn") == 1)

    DeltaTable.forPath(spark, "s3://freshcart-lake/silver/orders").alias("target") \
        .merge(latest.alias("source"), "target.order_id = source.order_id") \
        .whenMatchedUpdateAll().whenNotMatchedInsertAll().execute()

silver_stream = changes.writeStream.foreachBatch(merge_to_silver) \
    .option("checkpointLocation", "s3://checkpoints/orders_silver") \
    .trigger(processingTime="5 minutes").start()

ENABLING CHANGE DATA FEED ON A DELTA TABLE:
ALTER TABLE bronze.orders SET TBLPROPERTIES (delta.enableChangeDataFeed = true);
From this point, Delta records change type alongside each row change. CDF
adds storage overhead (~10-20% for typical workloads) — only enable it
when Change Data Feed is actively consumed.

FreshCart runs two separate systems: a Spark + S3 data lake (used by ML engineers) and a Snowflake warehouse (used by analysts). The daily ETL job that copies Silver tables from S3 to Snowflake costs $1,800/month in Snowflake credits and takes 4 hours. Analysts see data that is 4 hours stale. ML engineers and analysts have different versions of Silver — each team has found discrepancies. The CTO asks you to propose a consolidation.

Current state vs. target state

CURRENT STATE:
  S3 (lake):    raw/bronze/silver/gold layers, ML team reads silver via Spark
  Snowflake:    silver/gold tables (copy of lake silver, 4h stale),
                analyst team queries via Snowflake SQL
  ETL pipeline: copies silver from S3 to Snowflake daily (4h, $1,800/mo)
  Problem:      two copies, inconsistency, stale data, unnecessary cost

TARGET STATE:
  S3 (Delta Lake): raw/bronze/silver/gold layers — single copy
  Databricks SQL:  queries Silver/Gold Delta tables directly (replaces Snowflake)
  ML team:         reads same Delta tables via Spark (no change)
  Analyst team:    queries via Databricks SQL (new tool, same SQL)
  ETL pipeline:    eliminated

Migration phases 1-2 — enabling Delta Lake, dual-write validation

PHASE 1 (Weeks 1-4): Enable Delta Lake on existing S3 tables
  - Convert Silver/Gold S3 Parquet tables to Delta Lake format:
    spark.read.parquet("s3://freshcart-lake/silver/orders") \
        .write.format("delta").save("s3://freshcart-lake/silver_delta/orders")
  - Validate: Delta and Parquet versions produce identical query results
  - Set up Unity Catalog: register Delta tables in Unity Catalog
  - Create a Databricks SQL Warehouse for analysts

PHASE 2 (Weeks 5-8): Dual-write period
  - Silver pipeline writes to BOTH S3 Parquet AND Delta simultaneously
  - Analysts get Databricks SQL access to Delta tables, run queries in
    parallel against Snowflake (old) and Databricks SQL (new)
  - Compare results: if Databricks SQL matches Snowflake → validation passed
  - ML team switches to Delta tables: verify training pipelines unchanged

Migration phases 3-4 — cutover, then optimisation

PHASE 3 (Weeks 9-12): Cutover
  - Stop writing to S3 Parquet (keep Delta as sole format)
  - Stop running the ETL copy to Snowflake
  - Decommission Snowflake warehouse (or retain for special use cases)
  - Migrate all dbt models to run against Delta tables (dbt Databricks adapter)
  - Redirect all analyst tools (Metabase) to the Databricks SQL endpoint

PHASE 4 (Week 13+): Optimise
  - Add CLUSTER BY on Delta tables for analyst query patterns
  - Tune Databricks SQL warehouse sizes (X-Small for dashboards, Medium for ad-hoc)
  - Configure auto-suspend appropriately
  - Enable Unity Catalog lineage and access control

EXPECTED OUTCOMES:
ETL cost eliminated:               -$1,800/month
Snowflake compute eliminated:      -$3,200/month (warehouse credits)
Snowflake storage eliminated:      -$800/month
Databricks SQL new cost:           +$1,600/month (smaller warehouse, less data)
Net saving:                        $4,200/month ($50,400/year)

Data freshness:                    4 hours → 15 minutes (pipeline interval)
ML/analyst consistency:            guaranteed (one copy)
Operational pipelines to maintain: from 2 (lake + ETL) to 1 (lake only)