This exercise builds Stage 4 and Stage 5 of the capstone pipeline: Gold mart construction, Athena-style SQL analytics, FinOps cost tagging, and Spot cost estimation. Using DuckDB SQL to simulate Athena and BigQuery query patterns from Modules 2 and 3, you will build the `gold_bowler_season_stats` mart from Silver, run three analytical queries (economy by phase, powerplay vs death economy comparison, wicket-taking in pressure phases), validate all pipeline resource tags, estimate the monthly Spot vs On-Demand cost, and run the complete integration test that asserts every correctness invariant simultaneously.
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 Gold layer is the consumer-facing output of the lakehouse pipeline. The Athena-style SQL queries simulate the analytical workload that BI tools, the fantasy platform, and the commentary team would run against the Gold layer in production. The integration test brings together all assertions from Lessons 32, 33, and 34 into a single verification step — every assertion must pass before the capstone is considered complete and ready for submission.
Step 1 — Gold Mart and Athena SQL Analytics
Build the Gold bowler season stats mart from Silver data using DuckDB SQL, write it to the Gold S3 path with an Iceberg overwrite snapshot, and run three Athena-style analytical queries: (1) economy rate by phase for each bowler, (2) comparison of powerplay vs death economy, and (3) identification of bowlers who took wickets in the highest-pressure phase. Assert delivery conservation (Gold deliveries sum equals Silver row count), all economy rates are non-negative, and the Iceberg snapshot correctly records the Gold overwrite operation.
Analogy🏏Cricket
🏏 Think of it like cricket: building the Gold mart is compiling the season's headline bowler table from the certified Silver ball-by-ball record, then letting analysts interrogate it. Just as a statistician rolls up every delivery a bowler sent down into a single season line — deliveries, runs conceded, wickets and economy — the DuckDB Gold SQL aggregates Silver into `gold_bowler_season_stats`, written to the Gold path with an Iceberg overwrite snapshot like publishing the definitive edition over any draft. Then, just as pundits mine that table with sharp questions — each bowler's economy split by innings phase, powerplay versus death comparisons, and who took wickets in the highest-pressure overs — the three Athena-style queries answer exactly those. Crucially, just as an auditor confirms the sum of a bowler's per-phase deliveries still equals his total balls bowled, the code asserts Gold delivery counts sum back to the Silver row count and every economy is non-negative. The payoff: a trustworthy, analytics-ready season table whose totals provably reconcile with the certified source.
python
# capstone_gold_analytics.py — Step 1: Gold mart and Athena SQL analytics
import duckdb
import pandas as pd
import numpy as np
np.random.seed(42)
con = duckdb.connect(":memory:")
con.register("silver_deliveries", silver_with_new_col)
# ── Gold mart: bowler season statistics ───────────────────────────────────────
GOLD_SQL = """
SELECT
bowler,
match_id,
COUNT(*) AS deliveries,
ROUND(COUNT(*) / 6.0, 1) AS overs,
SUM(runs_scored) AS runs_conceded,
SUM(CASE WHEN is_wicket THEN 1 ELSE 0 END) AS wickets,
SUM(CASE WHEN is_boundary THEN 1 ELSE 0 END) AS boundaries_conceded,
ROUND(SUM(runs_scored) / NULLIF(COUNT(*)/6.0, 0), 2) AS economy,
SUM(CASE WHEN phase = 'powerplay' THEN runs_scored ELSE 0 END) AS powerplay_runs,
SUM(CASE WHEN phase = 'death' THEN runs_scored ELSE 0 END) AS death_runs,
SUM(CASE WHEN phase = 'powerplay' AND is_wicket THEN 1 ELSE 0 END) AS powerplay_wickets,
SUM(CASE WHEN phase = 'death' AND is_wicket THEN 1 ELSE 0 END) AS death_wickets
FROM silver_deliveries
GROUP BY bowler, match_id
ORDER BY economy
"""
gold_df = con.execute(GOLD_SQL).df()
GOLD_PATH = "s3://ipl-lakehouse-bronze-prod/gold/bowler_stats/year=2024/month=04/day=20/gold.parquet"
S3[GOLD_PATH] = gold_df.copy()
snap_gold = iceberg_overwrite_snapshot("gold.bowler_stats", GOLD_PATH, len(gold_df))
# ── Conservation assertion ────────────────────────────────────────────────────
gold_delivery_total = int(gold_df["deliveries"].sum())
silver_row_count = len(silver_with_new_col)
assert gold_delivery_total == silver_row_count, \
f"Conservation: gold={gold_delivery_total} != silver={silver_row_count}"
assert gold_df["economy"].notna().all()
assert (gold_df["economy"] >= 0).all()
print(f" Gold: {len(gold_df)} bowler-match records")
print(f" Delivery conservation: {gold_delivery_total} == {silver_row_count} ✓")
print(f" Iceberg snapshot {snap_gold['snapshot_id']}: overwrite ✓")
# ── Athena-style SQL analytics ────────────────────────────────────────────────
con.register("gold_bowler_stats", gold_df)
# Q1: Economy by phase (Athena/BigQuery analytics query)
q1 = con.execute("""
SELECT bowler,
ROUND(powerplay_runs / NULLIF(powerplay_wickets, 0), 2) AS pp_runs_per_wicket,
ROUND(death_runs / NULLIF(death_wickets, 0), 2) AS death_runs_per_wicket,
economy
FROM gold_bowler_stats
ORDER BY economy
""").df()
print(f" Q1 (economy by phase): {len(q1)} rows ✓")
# Q2: Powerplay vs death economy comparison
q2 = con.execute("""
SELECT bowler,
ROUND(SUM(powerplay_runs) / NULLIF(SUM(CASE WHEN powerplay_runs > 0 THEN 1 ELSE 0 END) * 1.0, 0), 2) AS avg_powerplay_economy,
economy AS overall_economy
FROM gold_bowler_stats
GROUP BY bowler, economy
ORDER BY economy
""").df()
print(f" Q2 (powerplay vs overall economy): {len(q2)} rows ✓")
# Q3: Pressure phase wicket takers (death overs)
q3 = con.execute("""
SELECT bowler, SUM(death_wickets) AS total_death_wickets
FROM gold_bowler_stats
GROUP BY bowler
HAVING SUM(death_wickets) > 0
ORDER BY total_death_wickets DESC
""").df()
print(f" Q3 (death wicket takers): {len(q3)} bowlers with death wickets ✓")
print("Step 1 ✓: Gold mart and Athena SQL analytics complete")Step 2 — FinOps Validation and Full Integration Test
Validate the FinOps tags on all five pipeline resources (VPC, two S3 paths, Glue role, Lambda function), estimate the monthly pipeline cost with and without Spot instances and verify the Spot saving exceeds 60%, then run the complete integration test that asserts all nine correctness invariants simultaneously. A pass on all nine assertions marks the pipeline as submission-ready.
Analogy🏏Cricket
🏏 Think of it like cricket: this final step is the pre-publication audit and the referee's sign-off sheet that must be fully ticked before the season statistics go to press. Just as the finance officer confirms every ground asset — the stands, the floodlights, the equipment stores — carries its correct cost label before the books are closed, the FinOps check validates tags on all five pipeline resources (VPC, both S3 paths, Glue role, Lambda). Just as the franchise weighs fielding a full-price squad against a value-priced one and confirms the saving justifies the choice, the cost estimate compares On-Demand against Spot and verifies the Spot saving clears 60%. Then, just as a match result is only declared official once the referee's checklist has every single box ticked — no partial certification — the integration test asserts all nine correctness invariants simultaneously, and only a clean sweep marks the pipeline submission-ready. The payoff: cost accountability plus an all-or-nothing final gate, so the capstone ships only when every guarantee holds at once.
python
# capstone_gold_analytics.py — Step 2: FinOps and full integration test
print("\n" + "="*60)
print("Course 5 Capstone — Full Integration Test")
print("="*60)
final_assertions = []
# ── FinOps: validate tags on all resources ────────────────────────────────────
PIPELINE_RESOURCES = {
"vpc": IPL_INFRA["vpc"]["tags"],
"s3_bronze": IPL_INFRA["s3_bronze"]["tags"],
"glue_role": IPL_INFRA["glue_role"]["tags"],
"lambda_fn": {"Team":"data-eng","Project":"ipl-lakehouse","Environment":"prod",
"Pipeline":"capstone","Owner":"[email protected]","ManagedBy":"Terraform"},
"gold_table": {"Team":"data-eng","Project":"ipl-lakehouse","Environment":"prod",
"Pipeline":"capstone","Owner":"[email protected]","ManagedBy":"Terraform"},
}
all_tags_valid = all(
len(REQUIRED_TAGS - set(tags.keys())) == 0
for tags in PIPELINE_RESOURCES.values()
)
final_assertions.append(("FinOps: all resources have required tags", all_tags_valid))
# ── Cost estimation ────────────────────────────────────────────────────────────
ON_DEMAND_HR = 0.192 # m5.xlarge
SPOT_DISCOUNT = 0.70
DURATION_HR = 0.5
WORKERS = 4
cost_on_demand = ON_DEMAND_HR * WORKERS * DURATION_HR
cost_spot = cost_on_demand * (1 - SPOT_DISCOUNT)
spot_saving = 1 - cost_spot / cost_on_demand
final_assertions.append((f"FinOps: Spot saving ≥ 60% ({spot_saving:.0%})", spot_saving >= 0.60))
# ── Integration test: all correctness invariants ──────────────────────────────
# A1: Terraform validation
terraform_ok = len(validate_cidrs(IPL_INFRA) + validate_tags(IPL_INFRA) +
validate_s3_security(IPL_INFRA) + validate_sg_no_public_db(IPL_INFRA)) == 0
final_assertions.append(("Stage 1: Terraform validation passed", terraform_ok))
# A2: Lambda idempotency
final_assertions.append(("Stage 2: Lambda idempotency (1 run per file)", len(PIPELINE_REGISTRY) == 1))
# A3: Bronze immutability
bronze_immutable = BRONZE_PATH in S3
final_assertions.append(("Stage 2: Bronze S3 path exists", bronze_immutable))
bronze_row_count = len(S3[BRONZE_PATH])
final_assertions.append((f"Stage 2: Bronze has {bronze_row_count} rows (raw)", bronze_row_count == 252))
# A4: Silver quality
silver_row_count = len(silver_with_new_col)
final_assertions.append((f"Stage 3: Silver {silver_row_count} < Bronze {bronze_row_count}",
silver_row_count < bronze_row_count))
silver_quality_ok = (silver_with_new_col["bowler"].notna().all() and
silver_with_new_col["runs_scored"].between(0, 6).all())
final_assertions.append(("Stage 3: Silver quality gate passed", silver_quality_ok))
# A5: Gold conservation
final_assertions.append((f"Stage 4: Gold delivery conservation ({gold_delivery_total}=={silver_row_count})",
gold_delivery_total == silver_row_count))
# A6: Iceberg snapshots
snap_ops = [s["operation"] for s in ICEBERG_SNAPSHOTS]
final_assertions.append((f"Stage 4: 4 Iceberg snapshots (1 append + 3 overwrite)",
len(ICEBERG_SNAPSHOTS) == 4 and
snap_ops.count("append") == 1 and
snap_ops.count("overwrite") == 3))
# ── Print results ──────────────────────────────────────────────────────────────
all_pass = True
for name, ok in final_assertions:
print(f" {'✓' if ok else '✗'} {name}")
if not ok: all_pass = False
print()
if all_pass:
print(f"ALL {len(final_assertions)} ASSERTIONS PASSED")
print("Course 5 Capstone is submission-ready. ✓")
else:
failed = [n for n, ok in final_assertions if not ok]
raise AssertionError(f"FAILED: {failed}")