Add dbt Transformations and Tests — Data Pipeline Orchestration

This exercise implements Stage 3 of the capstone pipeline: the dbt transformation layer that builds on the staged delivery data from Stage 1. Using DuckDB as the in-memory warehouse, you will implement all three model layers — staging, intermediate, and mart — with the economy rate macro, an incremental fact table with watermark-based filtering, and a complete `schema.yml` with six generic tests. After building the models, the exercise runs the simulated dbt tests and verifies that the source freshness check correctly identifies stale data before the dbt build proceeds.

Analogy🏏Cricket

🏏 Think of it like cricket: Migrating from Airflow to Prefect is like the same bowling coach shifting from traditional Test cricket notation to a modern T20 analytics dashboard — the underlying ball-by-ball data (the business logic) is exactly the same. What changes is how the data is recorded, displayed, and acted upon. The yorker that Bumrah bowls in over 20 is identical whether it is recorded in the old scorebook (Airflow DAG file) or the new analytics platform (Prefect flow). The migration is a transcription exercise, not a strategy change — and a wise coach verifies that the runs, wickets, and economies match exactly between the old and new system before decommissioning the scorebook. That verification step is the whole heart of the migration: because the yorker is unchanged, the only honest test is to run the same over through both systems and confirm the recorded runs, wickets and economies match to the last digit before the old scorebook is thrown away. Rushing to burn the scorebook the moment the shiny dashboard lights up is how teams lose a season of records to a silent transcription slip. The coach keeps both systems running in parallel for a while, reconciles their outputs ball by ball, and only when every figure agrees does he trust the new dashboard alone — a transcription is only complete when you have proven nothing was lost in the copying.

The connection between the Airflow DAG layer (Stage 1–2) and the dbt layer (Stage 3) is the staging area: the Airflow pipeline writes the validated delivery records to a staging table, and dbt reads from that staging table as its source. This boundary is the ELT pattern's clean separation: Airflow handles extraction and loading, dbt handles transformation. In production, the dbt run is triggered by a `BashOperator` in the Airflow DAG after the quality gate passes; in this exercise, the dbt SQL is executed directly in DuckDB.

Step 1 — Staging and Intermediate Models

Load the validated delivery records from Stage 2 into DuckDB as the raw source, implement the staging model that renames and casts columns, implement the intermediate model that joins deliveries with match metadata, and verify the row count and column schema at each layer. Assert that no rows are lost in the staging filter (all delivery_ids are non-null) and that no rows are lost in the join (all deliveries have non-null venue after left-joining the matches source).

Analogy🏏Cricket

🏏 Think of it like cricket: The staging-then-intermediate-then-mart progression in this exercise is the statistics department refining a rough scorer's sheet into a broadcast-ready season table in disciplined steps, never in one messy leap. The staging model is the first desk: it takes the validated raw delivery records loaded into DuckDB and does nothing clever, only renames the columns and casts the types so the sheet is tidy and consistent — the same match, just legible. The intermediate model is the middle desk where deliveries are enriched and joined, partnerships and contexts assembled, but still not yet the final figure. Only the mart model performs the headline aggregation — the bowler's season economy rate, computed once via the shared macro formula — producing the number that actually goes on air. Each desk trusts the one before it and touches the raw feed only through the staging model, exactly as a good stats department never lets an unprocessed scorer's scrawl jump straight onto the broadcast graphic. Refine in stages, and every layer is simpler and checkable than a single tangled query would ever be.

python

# capstone_dbt_transform.py — Step 1: Staging and intermediate models
import duckdb
import pandas as pd
import numpy as np
from datetime import datetime, timezone, timedelta

np.random.seed(42)

con = duckdb.connect(":memory:")

# ── Load validated delivery data (output from Stage 2) ────────────────────────
deliveries_validated = pd.DataFrame([
    {"delivery_id": mid * 1000 + i,
     "match_id":    mid,
     "runs":        int(np.random.choice([0,1,2,4,6])),
     "is_wicket":   bool(np.random.random() < 0.05),
     "bowler":      np.random.choice(["Bumrah","Shami","Hardik","Ashwin"]),
     "batter":      np.random.choice(["Rohit","Kohli","Gill","Dhoni"]),
     "batting_team": "Mumbai Indians" if mid == 10001 else
                    "RCB"            if mid == 10002 else "KKR",
     "over_number":  (i // 6) + 1,
     "ball_number":  (i % 6) + 1,
     "extras":       0,
     "_loaded_at":   (datetime.now(timezone.utc) - timedelta(hours=1)).isoformat(),
    }
    for mid in [10001, 10002, 10003]
    for i in range(120)
])

matches_raw = pd.DataFrame({
    "match_id":  [10001, 10002, 10003],
    "season":    [2024, 2024, 2024],
    "match_date":["2024-04-20", "2024-04-20", "2024-04-20"],
    "venue":     ["Wankhede", "Chinnaswamy", "Eden Gardens"],
    "home_team": ["Mumbai Indians", "RCB", "KKR"],
    "away_team": ["CSK", "KKR", "SRH"],
})

con.register("raw_deliveries", deliveries_validated)
con.register("raw_matches",    matches_raw)

# ── Staging model: stg_ipl_deliveries ─────────────────────────────────────────
stg_sql = """
SELECT
    delivery_id,
    match_id,
    over_number                                 AS over,
    ball_number                                 AS ball,
    bowler,
    batter,
    batting_team,
    CAST(runs AS INTEGER)                       AS runs,
    CAST(extras AS INTEGER)                     AS extras,
    CAST(is_wicket AS BOOLEAN)                  AS is_wicket,
    CASE
        WHEN over_number <= 6  THEN 'powerplay'
        WHEN over_number <= 15 THEN 'middle'
        ELSE 'death'
    END                                         AS phase,
    CAST(runs AS INTEGER) >= 4                  AS is_boundary,
    CAST(_loaded_at AS TIMESTAMP WITH TIME ZONE) AS loaded_at
FROM raw_deliveries
WHERE delivery_id IS NOT NULL
"""
con.execute(f"CREATE OR REPLACE VIEW stg_ipl_deliveries AS {stg_sql}")
stg_df = con.execute("SELECT * FROM stg_ipl_deliveries").df()
assert len(stg_df) == 360, f"Expected 360, got {len(stg_df)}"
assert "phase" in stg_df.columns
assert stg_df["delivery_id"].notna().all()
print(f"  stg_ipl_deliveries: {len(stg_df)} rows ✓")

# ── Staging model: stg_ipl_matches ────────────────────────────────────────────
con.execute("""
    CREATE OR REPLACE VIEW stg_ipl_matches AS
    SELECT match_id, season, CAST(match_date AS DATE) AS match_date,
           venue, home_team, away_team
    FROM raw_matches
""")

# ── Intermediate model: int_delivery_enriched ─────────────────────────────────
int_sql = """
SELECT
    d.*,
    m.season, m.match_date, m.venue, m.home_team, m.away_team
FROM stg_ipl_deliveries d
LEFT JOIN stg_ipl_matches m USING (match_id)
"""
con.execute(f"CREATE OR REPLACE VIEW int_delivery_enriched AS {int_sql}")
int_df = con.execute("SELECT * FROM int_delivery_enriched").df()
assert len(int_df) == 360
assert int_df["venue"].notna().all(), "Null venues after join"
assert set(int_df["phase"].unique()) <= {"powerplay","middle","death"}
print(f"  int_delivery_enriched: {len(int_df)} rows, all venues populated ✓")

# ── Source freshness simulation ────────────────────────────────────────────────
from datetime import datetime, timezone, timedelta
latest_ts  = pd.to_datetime(stg_df["loaded_at"]).max()
if latest_ts.tzinfo is None:
    latest_ts = latest_ts.tz_localize("UTC")
age_hours  = (datetime.now(timezone.utc) - latest_ts).total_seconds() / 3600
freshness_ok = age_hours <= 24
print(f"  Source freshness: {age_hours:.2f}h (max 24h) — {'OK' if freshness_ok else 'STALE'} ✓")
assert freshness_ok, f"Source data is stale: {age_hours:.1f}h old"
print("Step 1 ✓: staging and intermediate models complete")

Step 2 — Mart Model, Incremental Simulation and dbt Tests

Build the `fct_ipl_bowler_season_stats` mart with the economy rate macro inlined, simulate the incremental run by adding 30 new deliveries for a fourth match and verifying the mart updates correctly, then run all six simulated dbt tests and confirm all pass for clean data. Inject a bad row to confirm the economy range test fires correctly, and verify the complete dbt model hierarchy with a row count conservation assertion: total deliveries across all bowlers in the mart equals total deliveries in the intermediate model.

Analogy🏏Cricket

🏏 Think of it like cricket: The mart is the official season table, built here with the economy-rate macro inlined. The incremental simulation is the mid-tournament update when a fourth match's thirty new deliveries arrive: rather than re-tabulating the whole season, the statistician folds the new figures into the running totals and re-sums, then proves nothing was lost by checking that total deliveries in the mart exactly equal the delivery count in the intermediate model. The six dbt tests are the audit checklist — bowler and season not null, economy in range, no duplicate bowler. Just as an auditor never trusts a check that has only seen clean scorecards, this step injects an impossible economy of 48 and confirms the range test fires on exactly that one row. Practising the incremental fold and the deliberate bad row is what proves both that the update conserves every ball and that each test genuinely catches the violation it guards against.

python

# capstone_dbt_transform.py — Step 2: Mart, incremental, dbt tests
import numpy as np
np.random.seed(99)

# ── Mart model: fct_ipl_bowler_season_stats (economy rate macro inlined) ──────
fct_sql = """
SELECT
    bowler,
    season,
    COUNT(*)                                                    AS deliveries,
    ROUND(COUNT(*) / 6.0, 1)                                    AS overs,
    SUM(runs + extras)                                          AS runs_conceded,
    SUM(CASE WHEN is_wicket THEN 1 ELSE 0 END)                  AS wickets,
    ROUND(SUM(runs+extras) / NULLIF(COUNT(*) / 6.0, 0), 2)     AS economy
FROM int_delivery_enriched
GROUP BY bowler, season
"""
con.execute(f"CREATE OR REPLACE TABLE fct_ipl_bowler_season_stats AS {fct_sql}")
fct_df = con.execute("SELECT * FROM fct_ipl_bowler_season_stats").df()
assert len(fct_df) > 0
assert set(fct_df.columns) >= {"bowler","season","deliveries","economy","wickets"}
print(f"  fct_ipl_bowler_season_stats: {len(fct_df)} bowler records ✓")

# ── Incremental simulation ─────────────────────────────────────────────────────
new_deliveries = pd.DataFrame([
    {"delivery_id": 10004*1000+i, "match_id": 10004,
     "runs": int(np.random.choice([0,1,2,4,6])),
     "is_wicket": bool(np.random.random() < 0.05),
     "bowler": np.random.choice(["Chahal","Rashid","Bumrah"]),
     "batter": np.random.choice(["Rohit","Kohli"]),
     "batting_team": "MI",
     "over_number": (i//6)+1, "ball_number": (i%6)+1, "extras": 0,
     "_loaded_at": datetime.now(timezone.utc).isoformat(),
    }
    for i in range(30)
])
new_matches = pd.DataFrame([{
    "match_id": 10004, "season": 2024, "match_date": "2024-04-21",
    "venue": "Wankhede", "home_team": "MI", "away_team": "CSK",
}])

# Re-register sources with new data appended
all_del = pd.concat([deliveries_validated, new_deliveries], ignore_index=True)
all_mat = pd.concat([matches_raw, new_matches], ignore_index=True)
con.register("raw_deliveries", all_del)
con.register("raw_matches",    all_mat)

# Incremental: only process rows newer than current max loaded_at
current_max_ts = "2024-04-20T00:00:00+00:00"
con.execute(f"""
    CREATE OR REPLACE TABLE fct_ipl_bowler_season_stats AS
    WITH incremental AS (
        SELECT * FROM int_delivery_enriched
        WHERE loaded_at > TIMESTAMPTZ '{current_max_ts}'
    ),
    merged AS (
        SELECT bowler, season,
            SUM(deliveries) AS deliveries, SUM(overs) AS overs,
            SUM(runs_conceded) AS runs_conceded, SUM(wickets) AS wickets
        FROM (
            SELECT bowler, season, deliveries, overs, runs_conceded, wickets
            FROM fct_ipl_bowler_season_stats
            UNION ALL
            SELECT bowler, season, COUNT(*) AS deliveries,
                ROUND(COUNT(*)/6.0,1) AS overs,
                SUM(runs+extras) AS runs_conceded,
                SUM(CASE WHEN is_wicket THEN 1 ELSE 0 END) AS wickets
            FROM incremental
            GROUP BY bowler, season
        )
        GROUP BY bowler, season
    )
    SELECT bowler, season, deliveries, overs, runs_conceded, wickets,
           ROUND(runs_conceded / NULLIF(deliveries / 6.0, 0), 2) AS economy
    FROM merged
""")
fct_after = con.execute("SELECT * FROM fct_ipl_bowler_season_stats").df()
assert len(fct_after) >= len(fct_df)
print(f"  After incremental: {len(fct_after)} bowler records (was {len(fct_df)}) ✓")

# ── Simulated dbt tests ────────────────────────────────────────────────────────
def not_null(df, col):    return df[df[col].isna()]
def unique(df, col):       return df[df.duplicated(subset=[col])]
def range_check(df, col, lo, hi): return df[~df[col].between(lo, hi)]
def not_null_or_empty(df, col): return df[df[col].isna() | (df[col].astype(str) == "")]

tests = [
    ("not_null bowler",          not_null(fct_after, "bowler")),
    ("not_null season",          not_null(fct_after, "season")),
    ("not_null economy",         not_null(fct_after, "economy")),
    ("not_null deliveries",      not_null(fct_after, "deliveries")),
    ("unique bowler",             unique(fct_after, "bowler")),
    ("economy in [0,36]",         range_check(fct_after, "economy", 0, 36)),
]
print("\ndbt test results:")
all_pass = True
for name, failing_rows in tests:
    status = "PASS" if len(failing_rows)==0 else f"FAIL ({len(failing_rows)} rows)"
    print(f"  {name:<35}: {status}")
    if len(failing_rows) > 0: all_pass = False
assert all_pass, "dbt tests failed on clean data"

# Inject bad economy to confirm test fires
bad  = pd.concat([fct_after, pd.DataFrame([{"bowler":"Bad","season":2024,
    "deliveries":1,"overs":0.2,"runs_conceded":8,"wickets":0,"economy":48.0}])],
    ignore_index=True)
fail = range_check(bad, "economy", 0, 36)
assert len(fail) == 1
print(f"  economy range test on bad data: FAIL ({len(fail)} row) ✓")

# Row count conservation
total_mart_deliveries = int(fct_after["deliveries"].sum())
total_int_deliveries  = len(all_del)  # includes new deliveries
assert total_mart_deliveries == total_int_deliveries, \
    f"Conservation fail: mart {total_mart_deliveries} ≠ int {total_int_deliveries}"
print(f"  Row conservation: {total_mart_deliveries} == {total_int_deliveries} ✓")
print("Step 2 ✓: mart model, incremental, dbt tests complete")

Warning: The row count conservation assertion — total deliveries in the mart equals total deliveries in the intermediate model — is the most important correctness check for any aggregation pipeline. It will fail if the GROUP BY introduces unexpected null grouping (a null bowler name groups all its rows under a single null key, producing fewer groups than expected) or if the incremental filter over-constrains the new data (missing some rows that should have been processed). Always run this assertion as the final test after both the full build and the incremental update.

Extension Challenge: Add a `dim_ipl_players` dimension table that deduplicates the bowler and batter names from the intermediate model, assigns each a surrogate `player_id`, and adds a `is_allrounder` boolean column for players appearing in both the bowling and batting fields. Write a custom singular test that verifies no player appears as both the bowler and batter in the same delivery — a physical impossibility in cricket. This adds the dimension table to the star schema and demonstrates how custom dbt tests encode domain-specific business rules that generic tests cannot express.