This exercise builds Stage 3 of the capstone pipeline: the Glue-style ETL transformation from Bronze to Silver. Using DuckDB SQL to simulate the Glue PySpark transformation from Module 2 Lesson 8, you will implement the full Silver transformation — deduplication, type casting, null rejection, quality gate, and phase/boundary enrichment — and write the output to the Silver S3 path with an Iceberg snapshot. The exercise also simulates a Glue Job Bookmark by tracking which Bronze files have been processed and implements a schema evolution step adding a `wicket_phase` column.
Analogy🏏Cricket
🏏 Think of it like cricket: This exercise is the IPL official statistics team building their daily automated processing pipeline — the complete workflow that takes raw ball-by-ball records from every ground and produces the certified statistics published on the official website by midnight. Stage 1 is the data catalogue check: verify that the incoming scorecards match the expected format before any processing begins. Stage 2 is the statistics calculation: joins with match metadata, derivation of over-level stats, economy rate computation. Stage 3 is the official record update: load the new statistics into the production database using the certified upsert protocol — delete the old version of today's record and insert the freshly computed one — so no match ever has two records in the official database.
The key correctness assertions for Silver are: row count is less than Bronze (duplicates and nulls removed), all Silver rows pass the quality gate (runs in [0,6], bowler not null), the Iceberg snapshot records an overwrite operation with the correct row count, and the schema evolution step adds `is_boundary` to existing rows with the correct default value. A moto-based S3 interaction test verifies that the Bronze path read and Silver path write work correctly using the AWS SDK mock.
Step 1 — Glue ETL Transformation and Iceberg Silver Snapshot
Read Bronze data from the simulated S3 store, apply the full Silver transformation SQL (deduplication + type casting + quality gate + enrichment) using DuckDB, write Silver to a new S3 path with Iceberg overwrite snapshot, implement the Glue Job Bookmark simulation (track processed Bronze files), and verify the Silver Iceberg snapshot records an overwrite operation. Assert Silver row count is less than Bronze, all quality checks pass, and the Job Bookmark correctly marks the Bronze file as processed.
Analogy🏏Cricket
🏏 Think of it like cricket: this step is the statistics desk reading the sealed Bronze scorecard, producing the certified clean version, and logging that this match sheet is now processed so it is never re-run by mistake. Just as a statistician removes the duplicate ball entries, discards the lines with a missing bowler, and rewrites every scrawled string figure into a proper number before the record is certified, the DuckDB Silver SQL deduplicates, casts types and applies the quality gate — asserting the Silver row count comes out strictly below Bronze because the bad balls were dropped. Just as the certified sheet replaces, rather than appends to, any earlier draft, the Silver write records an Iceberg overwrite snapshot with the correct row count. Just as the desk ticks off each raw scorecard in its processed-log so the same match is never certified twice, the Glue Job Bookmark marks the Bronze file as done. The payoff: a clean, quality-gated Silver layer produced exactly once per Bronze file, with a snapshot proving what was overwritten.
python
# capstone_silver_transform.py — Step 1: Glue ETL and Iceberg Silver snapshot
import duckdb
import pandas as pd
import numpy as np
from datetime import datetime, timezone
# Reuse S3, ICEBERG_SNAPSHOTS from Lesson 32 (in full pipeline: passed as state)
# For standalone: reinitialise
np.random.seed(42)
# ── Read from Bronze S3 ───────────────────────────────────────────────────────
bronze_df = S3[BRONZE_PATH].copy()
assert len(bronze_df) == 252, f"Expected 252 bronze rows, got {len(bronze_df)}"
con = duckdb.connect(":memory:")
con.register("bronze_raw", bronze_df)
# ── Silver transformation (Glue PySpark equivalent in DuckDB SQL) ─────────────
SILVER_SQL = """
WITH deduplicated AS (
SELECT *,
ROW_NUMBER() OVER (PARTITION BY delivery_id ORDER BY _ingested_at) AS _rn
FROM bronze_raw
),
type_cast AS (
SELECT
CAST(delivery_id AS BIGINT) AS delivery_id,
CAST(match_id AS INTEGER) AS match_id,
CAST(over AS INTEGER) AS over,
CAST(runs_scored AS INTEGER) AS runs_scored,
CAST(is_wicket AS BOOLEAN) AS is_wicket,
bowler,
batter,
CAST(_ingested_at AS TIMESTAMPTZ) AS ingested_at
FROM deduplicated
WHERE _rn = 1
AND bowler IS NOT NULL
AND CAST(runs_scored AS INTEGER) BETWEEN 0 AND 6
),
enriched AS (
SELECT *,
CASE
WHEN over <= 6 THEN 'powerplay'
WHEN over <= 15 THEN 'middle'
ELSE 'death'
END AS phase,
runs_scored >= 4 AS is_boundary
FROM type_cast
)
SELECT * FROM enriched
"""
silver_df = con.execute(SILVER_SQL).df()
# Write Silver to S3 (mutable — MERGE-based)
SILVER_PATH = "s3://ipl-lakehouse-bronze-prod/silver/deliveries/year=2024/month=04/day=20/silver.parquet"
S3[SILVER_PATH] = silver_df.copy()
# Iceberg overwrite snapshot for Silver
def iceberg_overwrite_snapshot(table: str, path: str, n_rows: int) -> dict:
snap = {
"snapshot_id": len(ICEBERG_SNAPSHOTS) + 1,
"operation": "overwrite",
"table": table,
"path": path,
"added_records": n_rows,
"deleted_records": 0,
"timestamp": datetime.now(timezone.utc).isoformat(),
}
ICEBERG_SNAPSHOTS.append(snap)
return snap
snap2 = iceberg_overwrite_snapshot("silver.deliveries", SILVER_PATH, len(silver_df))
# ── Assertions ────────────────────────────────────────────────────────────────
assert len(silver_df) < len(bronze_df), "Silver must be smaller than Bronze"
assert silver_df["bowler"].notna().all(), "No null bowlers in Silver"
assert silver_df["runs_scored"].between(0, 6).all(), "All runs in [0,6]"
assert "phase" in silver_df.columns
assert "is_boundary" in silver_df.columns
assert snap2["operation"] == "overwrite"
assert snap2["added_records"] == len(silver_df)
print(f" Silver: {len(silver_df)} rows (bronze={len(bronze_df)}, rejected={len(bronze_df)-len(silver_df)}) ✓")
print(f" Quality: no null bowlers, all runs in [0,6] ✓")
print(f" Iceberg snapshot {snap2['snapshot_id']}: overwrite, {snap2['added_records']} records ✓")
# ── Glue Job Bookmark simulation ──────────────────────────────────────────────
JOB_BOOKMARK: dict[str, str] = {} # source_path → processed_at
def bookmark_processed(path: str) -> None:
JOB_BOOKMARK[path] = datetime.now(timezone.utc).isoformat()
def is_already_processed(path: str) -> bool:
return path in JOB_BOOKMARK
bookmark_processed(BRONZE_PATH)
assert is_already_processed(BRONZE_PATH) == True
assert is_already_processed("other/path.parquet") == False
print(f" Job Bookmark: {BRONZE_PATH} marked as processed ✓")
print("Step 1 ✓: Glue ETL, Iceberg Silver snapshot, Job Bookmark complete")Step 2 — Schema Evolution and Idempotency
Simulate an Iceberg schema evolution event by adding a `wicket_phase` column to the Silver table — existing rows receive a default of `None`, new rows receive the computed phase value. Then run the full Silver transformation a second time for the same logical date and assert the Silver row count and aggregate runs sum are identical (idempotency). Verify the Iceberg snapshot count is correct: 3 total snapshots across both layers (1 Bronze append + 1 Silver overwrite + 1 Silver overwrite for the re-run).
Analogy🏏Cricket
🏏 Think of it like cricket: this step is the record book gaining a brand-new stat column mid-season and then proving that re-certifying the same day's play changes nothing. Just as the board can decide to start tracking which phase each wicket fell in and rule that every older entry simply carries a blank until recomputed, the Iceberg schema evolution adds a `wicket_phase` column where existing rows default to None and only new rows receive the computed value — an additive change that never breaks the sheets already filed. Then, just as a statistician re-running the exact same match's numbers must arrive at precisely the same totals or the record is untrustworthy, running the full Silver transformation a second time for the same logical date yields an identical Silver row count and identical runs-sum. Confirming the snapshot count is correct — three in total — is like checking the ledger logged each certification exactly once. The payoff: the schema can grow safely and the transformation is provably idempotent, so re-processing is always safe and never doubles the record.
python
# capstone_silver_transform.py — Step 2: Schema evolution and idempotency
import pandas as pd
import numpy as np
# ── Schema evolution: add wicket_phase column ─────────────────────────────────
silver_with_new_col = silver_df.copy()
silver_with_new_col["wicket_phase"] = silver_with_new_col.apply(
lambda r: r["phase"] if r["is_wicket"] else None,
axis=1,
)
# Old rows (without wickets in their original data) get None by default
assert "wicket_phase" in silver_with_new_col.columns
assert silver_with_new_col.loc[~silver_with_new_col["is_wicket"], "wicket_phase"].isna().all()
assert silver_with_new_col.loc[silver_with_new_col["is_wicket"], "wicket_phase"].notna().all()
print(f" Schema evolution: 'wicket_phase' added")
print(f" Wicket rows with phase: {silver_with_new_col['is_wicket'].sum()}")
print(f" Non-wicket rows with None: {(~silver_with_new_col['is_wicket']).sum()} ✓")
# Update Silver S3 path
S3[SILVER_PATH] = silver_with_new_col.copy()
# ── Idempotency: re-run Silver transformation for same logical date ────────────
con2 = duckdb.connect(":memory:")
con2.register("bronze_raw", bronze_df)
silver_run2 = con2.execute(SILVER_SQL).df()
# Add wicket_phase to run2 as well
silver_run2["wicket_phase"] = silver_run2.apply(
lambda r: r["phase"] if r["is_wicket"] else None, axis=1
)
# Row count idempotency
assert len(silver_run2) == len(silver_df), \
f"Idempotency failed: {len(silver_df)} → {len(silver_run2)}"
# Aggregate idempotency
runs_run1 = int(silver_df["runs_scored"].sum())
runs_run2 = int(silver_run2["runs_scored"].sum())
assert runs_run1 == runs_run2, f"Runs mismatch: {runs_run1} vs {runs_run2}"
# Record third Iceberg snapshot (Silver overwrite on re-run)
snap3 = iceberg_overwrite_snapshot("silver.deliveries", SILVER_PATH, len(silver_run2))
assert snap3["snapshot_id"] == 3
print(f" Silver idempotency: {len(silver_df)} == {len(silver_run2)} rows ✓")
print(f" Runs conservation: {runs_run1} == {runs_run2} ✓")
print(f" Total Iceberg snapshots: {len(ICEBERG_SNAPSHOTS)} (1 Bronze + 2 Silver) ✓")
print("Step 2 ✓: Schema evolution and idempotency complete")