Practice — Build dbt Models on Raw Postgres Data

This exercise builds a complete dbt project for the IPL analytics warehouse, implementing all five Module 5 concepts: a three-layer model structure (staging → intermediate → mart), an incremental fact table with a watermark-based `is_incremental()` filter, a reusable macro for economy rate, dbt tests in `schema.yml`, and a source freshness check. Because a live warehouse is not required for this exercise, all SQL is validated using DuckDB — an in-process SQL engine that supports the full SQL syntax and runs entirely in Python with no server required.

The exercise is structured in three steps. Step 1 implements the staging and intermediate models using DuckDB and verifies that the column transformations and join produce the expected output. Step 2 implements the incremental mart model, verifying that the first run processes all rows and the second run processes only new rows. Step 3 simulates dbt tests by running the compiled test SQL against the DuckDB result set and asserts that all tests pass for clean data and the correct tests fail when bad data is injected.

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.

Step 1 — Staging and Intermediate Models in DuckDB

Install DuckDB and create the raw source tables with the snake_case naming convention typical of source databases. Implement the staging SQL as Python string templates with Jinja-style substitutions, execute them against DuckDB, and verify the output schema and row counts. Then implement the intermediate join model and assert that the venue column is populated for all rows after the left join with the matches staging model.

Analogy🏏Cricket

🏏 Think of it like cricket: This step is the analytics team validating a new scoring methodology on the practice ground before it ever touches the official system. DuckDB is that practice net — an in-memory engine where real-shaped match data loads and queries instantly, with no production warehouse at risk. The staging model is the first desk that simply tidies the raw scorecard: snake_case source columns renamed to the team's convention, types cast correctly, obviously invalid records filtered out. The intermediate model is the join desk where each delivery is enriched with its match's venue. Just as a scorer double-checks that every ball still appears exactly once after enrichment, the assertion that venue is non-null on all rows after the left join proves no delivery was silently dropped. Practising the staging-then-join steps in a fast, disposable net is what lets the team catch a broken rename or a bad join before the methodology is ever trusted on the live statistics system.

python

# exercise_dbt_models.py — Step 1: Staging and intermediate models in DuckDB
# pip install duckdb pandas numpy
import duckdb
import pandas as pd
import numpy as np
from datetime import datetime, timezone, timedelta

np.random.seed(42)

# ── Create raw source data ────────────────────────────────────────────────────
con = duckdb.connect(":memory:")

# Raw deliveries (snake_case source naming convention)
deliveries_raw = pd.DataFrame({
    "delivery_id":      range(1, 241),
    "match_id":         [10001]*120 + [10002]*120,
    "over_number":      [(i//6)+1 for i in range(240)],
    "ball_number":      [(i%6)+1  for i in range(240)],
    "bowler_name":      np.random.choice(["Bumrah","Shami","Hardik","Ashwin"], 240),
    "batter_name":      np.random.choice(["Rohit","Kohli","Gill","Dhoni"], 240),
    "batting_team_code":np.random.choice(["MI","CSK"], 240),
    "runs_off_bat":      np.random.choice([0,1,2,4,6], 240),
    "extras":            np.zeros(240, dtype=int),
    "is_wicket":         np.random.choice([True,False], 240, p=[0.05,0.95]),
    "dismissal_kind":    None,
    "_loaded_at":        [datetime.now(timezone.utc) - timedelta(hours=1)]*240,
})

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

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

# ── Staging model: stg_ipl_deliveries ────────────────────────────────────────
stg_deliveries_sql = """
WITH source AS (
    SELECT * FROM raw_deliveries
),
renamed AS (
    SELECT
        delivery_id,
        match_id,
        over_number                             AS over,
        ball_number                             AS ball,
        bowler_name                             AS bowler,
        batter_name                             AS batter,
        batting_team_code                       AS batting_team,
        CAST(runs_off_bat AS INTEGER)           AS runs,
        CAST(extras AS INTEGER)                 AS extras,
        CAST(is_wicket AS BOOLEAN)              AS is_wicket,
        dismissal_kind                          AS dismissal_type,
        CASE
            WHEN over_number <= 6  THEN 'powerplay'
            WHEN over_number <= 15 THEN 'middle'
            ELSE 'death'
        END                                     AS phase,
        CAST(runs_off_bat AS INTEGER) >= 4      AS is_boundary,
        _loaded_at                              AS loaded_at
    FROM source
    WHERE delivery_id IS NOT NULL
)
SELECT * FROM renamed
"""

con.execute(f"CREATE OR REPLACE VIEW stg_ipl_deliveries AS {stg_deliveries_sql}")
stg_d = con.execute("SELECT * FROM stg_ipl_deliveries").df()

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

# 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()

# Assertions
assert len(stg_d)    == 240, f"Expected 240 rows, got {len(stg_d)}"
assert "phase"       in stg_d.columns
assert "is_boundary" in stg_d.columns
assert len(int_df)   == 240
assert int_df["venue"].notna().all(), "Null venues after join — check match_id FK"
assert set(int_df["phase"].unique()) <= {"powerplay", "middle", "death"}
print(f"Step 1 ✓: stg_ipl_deliveries({len(stg_d)} rows), int_delivery_enriched({len(int_df)} rows)")

Step 2 — Incremental Mart Model and dbt Tests

Build the `fct_ipl_bowler_season_stats` mart model with the economy rate macro inlined as a SQL expression, implement the incremental simulation by splitting the data into an initial batch and a second-run batch, and verify row count conservation. Then run the simulated dbt tests — `not_null`, `unique`, `accepted_range` for economy — against the mart output and inject a bad row to confirm the test catches the violation.

Analogy🏏Cricket

🏏 Think of it like cricket: The mart model is the official season table published on the site, and this step builds it with the economy-rate formula computed once via the shared macro. The incremental simulation is the mid-season update: rather than re-tabulating every match from ball one, the statistician takes the fresh batch of deliveries, folds it into the running totals, and confirms no ball was lost or double-counted through a row-count conservation check. The dbt tests are the verification officer's checklist — economy inside range, no duplicate bowler records, no null keys. Just as a good auditor never trusts a check that has only ever seen clean data, this step deliberately injects an impossible economy of 45 and confirms the accepted-range test actually fires and catches exactly that one row. Practising both the clean pass and the deliberate bad-row failure is what proves each test is genuinely written to catch violations rather than trivially passing.

python

# exercise_dbt_models.py — Step 2: Mart model, incremental simulation, dbt tests
from datetime import date, datetime, timezone, timedelta
import pandas as pd
import duckdb
import numpy as np

np.random.seed(99)

# ── Mart model: fct_ipl_bowler_season_stats ───────────────────────────────────
# Economy rate macro inlined: ROUND(runs / NULLIF(deliveries / 6.0, 0), 2)
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()

print(f"\nMart model: {len(fct_df)} bowler records")
fct_df_sorted = fct_df.sort_values("economy").reset_index(drop=True)
print(fct_df_sorted[["bowler","deliveries","overs","runs_conceded","economy"]].to_string())

# Incremental simulation: simulate a second run with new deliveries
new_deliveries = pd.DataFrame({
    "delivery_id":      range(241, 361),
    "match_id":         [10003]*120,
    "over":             [(i//6)+1 for i in range(120)],
    "ball":             [(i%6)+1  for i in range(120)],
    "bowler":           np.random.choice(["Bumrah","Shami","Chahal"], 120),
    "batter":           np.random.choice(["Rohit","Kohli"], 120),
    "batting_team":     "Mumbai Indians",
    "runs":             np.random.choice([0,1,2,4,6], 120),
    "extras":           np.zeros(120, dtype=int),
    "is_wicket":        np.random.choice([True,False], 120, p=[0.05,0.95]),
    "dismissal_type":   None,
    "phase":            np.random.choice(["powerplay","middle","death"], 120),
    "is_boundary":      np.random.choice([True,False], 120, p=[0.2,0.8]),
    "season":           2024,
    "match_date":       "2024-04-22",
    "venue":            "Eden Gardens",
    "home_team":        "KKR",
    "away_team":        "MI",
    "loaded_at":        [datetime.now(timezone.utc)]*120,
})

# Append new deliveries to int_delivery_enriched and merge into mart
new_stats = new_deliveries.groupby(["bowler","season"]).agg(
    deliveries=("delivery_id", "count"),
    runs_conceded=("runs", "sum"),
    wickets=("is_wicket", "sum"),
).reset_index()
new_stats["overs"]   = (new_stats["deliveries"] / 6).round(1)
new_stats["economy"] = (new_stats["runs_conceded"] / (new_stats["deliveries"]/6)).round(2)

# Simulate merge (dbt incremental with unique_key='bowler,season')
old_fct = fct_df.copy()
merged_fct = pd.concat([old_fct, new_stats], ignore_index=True)
merged_fct = merged_fct.groupby(["bowler","season"]).agg(
    deliveries=("deliveries", "sum"),
    overs=("overs", "sum"),
    runs_conceded=("runs_conceded", "sum"),
    wickets=("wickets", "sum"),
).reset_index()
merged_fct["economy"] = (merged_fct["runs_conceded"] / (merged_fct["deliveries"]/6)).round(2)

print(f"\nAfter incremental merge: {len(merged_fct)} bowler records")
assert len(merged_fct) >= len(fct_df)  # may grow if new bowlers appear

# ── Simulated dbt tests ───────────────────────────────────────────────────────
def dbt_not_null_test(df, col): return df[df[col].isna()]
def dbt_unique_test(df, col):   return df[df.duplicated(subset=[col])]
def dbt_range_test(df, col, lo, hi): return df[~df[col].between(lo, hi)]

tests = [
    ("not_null bowler",   dbt_not_null_test(merged_fct, "bowler")),
    ("unique bowler",     dbt_unique_test(merged_fct, "bowler")),
    ("economy in [0,36]", dbt_range_test(merged_fct, "economy", 0, 36)),
]
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:<30}: {status}")
    if len(failing_rows) > 0: all_pass = False
assert all_pass, "dbt tests failed on clean data"

# Inject a bad row: economy = 45 (impossible)
bad_row = pd.DataFrame([{"bowler":"BadBowler","season":2024,
                          "deliveries":1,"overs":0.2,
                          "runs_conceded":7,"wickets":0,"economy":45.0}])
bad_fct  = pd.concat([merged_fct, bad_row], ignore_index=True)
failing  = dbt_range_test(bad_fct, "economy", 0, 36)
assert len(failing) == 1, f"Expected 1 failing row, got {len(failing)}"
print(f"\n  economy range test on bad data: FAIL ({len(failing)} row) ✓")
print("All dbt test simulations passed. Exercise complete.")

Warning: DuckDB's `CAST(runs_off_bat AS INTEGER)` will succeed silently even if `runs_off_bat` contains floating-point values like `4.0` — it truncates to `4`. However, if the source column contains string values like `'N/A'` instead of null for missing data, the cast will raise a `ConversionException`. Always inspect the raw source data for non-null invalid values (empty strings, placeholder strings like 'N/A', sentinel values like -1 or 9999) before writing staging model casts, and use `TRY_CAST` with a `COALESCE` fallback for columns with uncertain source data quality.

Extension Challenge: Add a `dim_ipl_players` model that pivots the delivery data to produce one row per player with their batting and bowling statistics as separate columns, using a CASE expression to populate null values for pure batters (zero wickets) and pure bowlers (zero runs as a batter). Write the four generic dbt tests (`not_null player_id`, `unique player_id`, `accepted_range economy [0, 36]`, `relationships match_id`) as Python simulation functions and verify all pass on the clean mart output. This completes the star schema with a dimension table joined to the delivery fact table.