Capstone — Submit Orchestration Pipeline Project

This is the final submission for Course 4: Data Pipeline Orchestration. You have built a complete, production-patterned automated multi-stage data pipeline integrating every module of the course: DAGs, tasks, idempotency, and retry strategies; Airflow sensors, dynamic task mapping, branching, and MWAA deployment concepts; Prefect flows, blocks, and caching; data quality dimensions, GE checkpoints, dbt tests, lineage, and alerting; and a complete dbt transformation layer with three model layers, incremental materialisation, Jinja templating, and CI/CD integration. The integration test is the definitive submission criterion.

Before submitting, run the final integration test code block below. It executes all five pipeline stages sequentially and asserts every correctness invariant: sensor behaviour, dynamic extraction with retry, GE quality gate pass/fail routing, warehouse idempotency, dbt staging and mart row counts, conservation of delivery totals across model layers, tiered alert routing, lineage event structure, and end-of-run summary. The integration test is the definitive acceptance criterion — a submission where all assertions pass represents a production-patterned, end-to-end orchestration pipeline demonstrating mastery of every concept in this course.

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.

Final Integration Test

Run the complete integration test below. It re-executes the full pipeline end-to-end: sensor → dynamic extraction with retry → GE quality gate → dbt build → monitoring and lineage. Each stage asserts its own correctness, and the final summary prints a pass/fail line for every assertion before confirming submission readiness. All assertions must pass — a submission with any failing assertion will require resubmission after fixing the identified issue.

Analogy🏏Cricket

🏏 Think of it like cricket: The final integration test is the scoreboard at stumps re-reading the whole day's play in order — sensor, dynamic extraction with retry, quality gate, dbt build, monitoring and lineage — and printing a pass-or-fail line for every single invariant. Just as the closing scoreboard doesn't re-bowl a single delivery but confirms that every recorded run, wicket and economy is internally consistent, this test doesn't recompute each stage from scratch; it re-checks the results each stage already produced and asserts they all hold together. Each stage stands like a batsman who must post his own runs — the sensor polling at least twice, match 10002 retried exactly once, the warehouse stable across reruns, the mart conserving its rows. And just as a single missing run on the board signals an error somewhere specific, one failing assertion sends the submission back for a fix. Every assertion that passes is a run on the board; pass them all and the trophy is yours.

python

# capstone_final_integration_test.py
# Run this file to verify all pipeline stages before submitting
print("=" * 65)
print("Course 4 Capstone — Final Integration Test")
print("=" * 65)

final_assertions = []

# ── Stage 1: Sensor ─────────────────────────────────────────────────────────
print("\n[Stage 1] Sensor")
final_assertions.append(("Sensor succeeded",           sensor_result == True))
final_assertions.append(("Sensor polled >= 2 times",   poll_count[0] >= 2))
print(f"  sensor: {sensor_result}, polls: {poll_count[0]} ✓")

# ── Stage 2: Dynamic extraction ─────────────────────────────────────────────
print("\n[Stage 2] Dynamic Extraction")
final_assertions.append(("3 matches extracted",         len(extract_results) == 3))
final_assertions.append(("Match 10002 retried",          retry_counts[10002] == 2))
final_assertions.append(("Match 10001 no retry",         retry_counts[10001] == 1))
print(f"  matches={len(extract_results)}, 10002 retries={retry_counts[10002]} ✓")

# ── Stage 3: Quality gate ────────────────────────────────────────────────────
print("\n[Stage 3] GE Quality Gate")
final_assertions.append(("GE suite passed on clean data", ge_result.success))
final_assertions.append(("Branch target = load_to_warehouse", branch_target == "load_to_warehouse"))
final_assertions.append(("Bad data routes to DLQ",        branch_bad == "route_to_dlq"))
print(f"  ge_passed={ge_result.success}, branch={branch_target} ✓")

# ── Stage 4: Warehouse idempotency ──────────────────────────────────────────
print("\n[Stage 4] Warehouse Idempotency")
wh = len(warehouse_db.get("2024-04-20", []))
final_assertions.append(("Warehouse has 360 rows (run 1)",  wh_count_run1 == 360))
final_assertions.append(("Warehouse idempotent (run 1==2)", wh_count_run1 == wh_count_run2))
final_assertions.append(("DLQ route leaves warehouse unchanged", wh_after_bad == wh_count_run2))
print(f"  run1={wh_count_run1}, run2={wh_count_run2}, after_bad={wh_after_bad} ✓")

# ── Stage 5: dbt models ──────────────────────────────────────────────────────
print("\n[Stage 5] dbt Models")
final_assertions.append(("stg_ipl_deliveries: 360 rows",   len(stg_df) == 360))
final_assertions.append(("int_delivery_enriched: 360 rows",len(int_df) == 360))
final_assertions.append(("Mart has bowler records",         len(fct_after) > 0))
mart_cons = int(fct_after["deliveries"].sum())
final_assertions.append((f"Row conservation: mart={mart_cons}=={len(all_del)}",
                          mart_cons == len(all_del)))
print(f"  stg={len(stg_df)}, int={len(int_df)}, mart_rows={len(fct_after)}, conservation ✓")

# ── Stage 6: Lineage ─────────────────────────────────────────────────────────
print("\n[Stage 6] Lineage")
final_assertions.append(("5 pipeline stages have lineage events",
                          len(jobs_with_events) == 5))
final_assertions.append((f"Total lineage events = {len(stages)*2}",
                          len(lineage_log) == len(stages) * 2))
lineage_ok = all(
    e["job"]["name"] and len(e["inputs"]) >= 1 and len(e["outputs"]) >= 1
    for e in lineage_log
)
final_assertions.append(("All events have job, inputs, outputs", lineage_ok))
print(f"  stages={len(jobs_with_events)}, events={len(lineage_log)}, valid={lineage_ok} ✓")

# ── Stage 7: Alerting ────────────────────────────────────────────────────────
print("\n[Stage 7] Alerting")
pagerduty_alerts.clear(); slack_alerts.clear()
tiered_alert("critical", "Test critical", "test")
tiered_alert("warning",  "Test warning",  "test")
final_assertions.append(("Critical → PagerDuty", len(pagerduty_alerts) == 1))
final_assertions.append(("Warning → Slack",       len(slack_alerts) == 1))
print(f"  PD={len(pagerduty_alerts)}, Slack={len(slack_alerts)} ✓")

# ── Final summary ────────────────────────────────────────────────────────────
print("\n" + "=" * 65)
print("Assertion Results:")
all_pass = True
for name, ok in final_assertions:
    icon = "✓" if ok else "✗"
    print(f"  {icon} {name}")
    if not ok: all_pass = False

print()
if all_pass:
    print(f"ALL {len(final_assertions)} ASSERTIONS PASSED")
    print("Course 4 Capstone is submission-ready.")
else:
    failed = [n for n, ok in final_assertions if not ok]
    print(f"FAILED: {failed}")

Submission Checklist

Verify each item before submitting. Required items cause automatic deductions if absent; recommended items affect the architecture and code quality score. The checklist is ordered by pipeline stage — verify each stage before proceeding, as later integration test assertions depend on earlier stages producing correct output. The checklist covers all 22 items from sensor behaviour through lineage event structure.

Analogy🏏Cricket

🏏 Think of it like cricket: The submission checklist is the captain's pre-match team sheet and equipment check, worked through in strict order before walking out. Each item is a box that must be ticked — the sensor polling correctly, the warehouse holding exactly 360 rows, the venues non-null after the join, the six dbt tests passing, the lineage events all structurally complete — and the required items are the ones that, if missing, cost you the match before a ball is bowled, just as a missing pad or an unregistered player draws an automatic penalty. The list is ordered by pipeline stage on purpose, because a later check depends on an earlier stage having produced correct output, exactly as you confirm the openers are padded up before worrying about the middle order. Working methodically down all twenty-two items before submitting is what stops an avoidable, self-inflicted deduction from undoing an otherwise sound innings.

python

# Submission checklist — verify each before submitting

# [REQUIRED] Stage 1: Orchestration
# 1. Sensor polls >= 2 times and succeeds
# 2. Dynamic extract creates exactly 3 task instances for 3 match IDs
# 3. Transient failure on match 10002 is retried and resolved
# 4. GE quality gate passes on clean data, branches correctly

# [REQUIRED] Stage 2: Warehouse
# 5. Warehouse has exactly 360 rows after run 1
# 6. Warehouse idempotency: row count unchanged after run 2
# 7. DLQ routing: warehouse unchanged when bad data is injected
# 8. ALL_DONE cleanup task fires regardless of branch taken

# [REQUIRED] Stage 3: dbt Models
# 9.  stg_ipl_deliveries: 360 rows, 'phase' and 'is_boundary' columns present
# 10. int_delivery_enriched: 360 rows, all venues non-null after join
# 11. fct_ipl_bowler_season_stats: > 0 bowler records
# 12. Row conservation: SUM(deliveries) in mart == total rows in intermediate

# [REQUIRED] Stage 4: dbt Tests
# 13. All 6 dbt tests pass on clean mart data
# 14. Economy range test FAILS when a row with economy=48 is injected

# [REQUIRED] Stage 5: Monitoring and Lineage
# 15. 5 pipeline stages have emitted lineage events (10 events total: 5 START + 5 COMPLETE)
# 16. All lineage events have non-null job name, >= 1 input, >= 1 output
# 17. Critical alerts route to PagerDuty; warnings route to Slack
# 18. Full integration test passes: all assertions return True

# [RECOMMENDED] Code Quality
# 19. All alert and lineage calls are wrapped in try/except (best-effort)
# 20. All tasks use logical_date as the idempotency key for reads and writes
# 21. The economy rate formula is implemented as a consistent expression across all models
# 22. Source freshness check asserts data is < 24 hours old before dbt runs

print("Checklist complete. Submit project files to the SkillVeris capstone portal.")

What You Have Built: Over the six modules of Course 4, you have built a production-patterned data pipeline orchestration platform that covers the complete orchestration lifecycle: DAGs with idempotency and retry; Airflow sensors, dynamic task mapping, XComs, connections, variables, and MWAA deployment; Prefect flows, deployments, work pools, caching, blocks, and migration from Airflow; data quality dimensions, Great Expectations suites, dbt tests, OpenLineage lineage, and tiered alerting; and a complete dbt transformation layer with three model layers, incremental materialisation, Jinja templating, documented schema.yml, and GitHub Actions CI/CD. You are equipped to design, build, test, and operate production data pipelines at professional engineering standards.