Monitor Quality and Set Up Alerts — Data Pipeline Orchestration

This exercise implements Stage 4 and Stage 5 of the capstone pipeline: quality monitoring with tiered alerting, SLA tracking, and OpenLineage lineage emission for all five pipeline stages. The monitoring layer sits across the full pipeline and fires notifications based on quality gate results, SLA breaches, and pipeline completion. The lineage layer emits START and COMPLETE events for each stage with structured input and output dataset references. The exercise ends with a full integration test that verifies all assertions across Stages 1–5.

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 monitoring implementation uses the tiered alert router from Module 4 and the OpenLineage event emitter from Module 4 Lesson 22. All alert and lineage calls are wrapped in `try/except` blocks to ensure they are best-effort — pipeline failures are reported by the monitoring layer, not caused by it. The integration test asserts that all five pipeline stages have emitted lineage events with non-null job names, at least one input and one output, and that alert counts match expectations for each test scenario.

Step 1 — SLA Monitoring and Tiered Alerting

Implement the SLA monitor that tracks each pipeline stage's start time and fires a Slack warning if it exceeds the configured limit, the tiered alert router that sends critical quality failures to PagerDuty and warnings to Slack, and the end-of-run summary task. Run three test scenarios: a clean pipeline run (no alerts), a quality gate failure (PagerDuty critical), and an SLA breach (Slack warning). Assert the alert counts match expectations for each scenario.

Analogy🏏Cricket

🏏 Think of it like cricket: The per-stage SLA monitor in this exercise is the match referee holding a separate stopwatch on every phase of play, because each phase has its own fair time limit. Extraction, quality-checking, transformation, monitoring — the referee starts a fresh clock as each begins and, if any single phase overruns its allotted window, he raises a warning flag for that phase specifically, not for the match as a whole, the way an over that drags past its reasonable time draws attention even though no wicket has fallen. The tiered alert router is then the graded response to whatever the clocks and checks report: a genuinely dangerous breach goes straight to the senior official on the critical channel, while a minor overrun is a quiet note on the warning channel — matching the severity of the intervention to the severity of the problem so the serious signals are never drowned by the trivial. Independent stopwatches plus graded flags mean a slow-but-successful stage is caught early and routed sensibly, long before a merely-late pipeline silently misses the decision it was feeding.

python

# capstone_monitoring.py — Step 1: SLA monitoring and tiered alerting
from datetime import datetime, timezone, timedelta
import time

# ── Alert sinks ────────────────────────────────────────────────────────────────
pagerduty_alerts: list = []
slack_alerts:     list = []

def mock_pagerduty(title: str, details: str) -> None:
    pagerduty_alerts.append({"title": title, "details": details})
    print(f"  [PAGERDUTY] {title}")

def mock_slack(message: str, severity: str = "warning") -> None:
    slack_alerts.append({"message": message, "severity": severity})
    print(f"  [SLACK/{severity.upper()}] {message}")

def tiered_alert(severity: str, title: str, details: str) -> None:
    try:
        if severity == "critical":
            mock_pagerduty(title, details)
        else:
            mock_slack(details, severity)
    except Exception as exc:
        print(f"  Alert failed (non-blocking): {exc}")

# ── SLA Monitor ────────────────────────────────────────────────────────────────
class PipelineSLAMonitor:
    def __init__(self):
        self.stage_starts: dict[str, datetime] = {}
        self.stage_slas:   dict[str, float]    = {}

    def start_stage(self, stage: str, sla_minutes: float) -> None:
        self.stage_starts[stage] = datetime.now(timezone.utc)
        self.stage_slas[stage]   = sla_minutes

    def check_stage(self, stage: str) -> dict:
        if stage not in self.stage_starts:
            return {"breach": False, "elapsed_min": 0}
        elapsed = (
            datetime.now(timezone.utc) - self.stage_starts[stage]
        ).total_seconds() / 60
        breach  = elapsed > self.stage_slas.get(stage, 60)
        return {"stage": stage, "elapsed_min": round(elapsed, 3),
                "sla_min": self.stage_slas.get(stage), "breach": breach}

# ── End-of-run summary ────────────────────────────────────────────────────────
def send_run_summary(
    logical_date: str,
    matches:      int,
    rows_loaded:  int,
    dlq_count:    int,
    ge_passed:    bool,
    sla_breach:   bool,
) -> str:
    status  = "✅ SUCCESS" if ge_passed and not sla_breach else "⚠️ ISSUES"
    message = (
        f"{status} | IPL daily pipeline {logical_date}\n"
        f"  Matches: {matches} | Rows: {rows_loaded:,} | DLQ: {dlq_count}"
        f"{' | ⚠️ SLA breached' if sla_breach else ''}"
    )
    mock_slack(message, severity="info" if ge_passed else "warning")
    return message

# ── Test scenarios ─────────────────────────────────────────────────────────────
pagerduty_alerts.clear()
slack_alerts.clear()
sla_monitor = PipelineSLAMonitor()

# Scenario 1: clean run — only info summary
sla_monitor.start_stage("extract",  sla_minutes=5)
sla_monitor.start_stage("quality",  sla_minutes=2)
sla_monitor.start_stage("dbt_build",sla_minutes=15)

time.sleep(0.01)  # simulate fast run

for stage in ["extract", "quality", "dbt_build"]:
    check = sla_monitor.check_stage(stage)
    if check["breach"]:
        tiered_alert("warning", f"SLA breach: {stage}",
                     f"{stage} took {check['elapsed_min']}min (limit {check['sla_min']}min)")

summary1 = send_run_summary("2024-04-20", matches=3, rows_loaded=360,
                             dlq_count=0, ge_passed=True, sla_breach=False)
assert len(pagerduty_alerts) == 0
assert len(slack_alerts) == 1  # only summary
print(f"  Scenario 1 (clean run): {len(slack_alerts)} Slack, {len(pagerduty_alerts)} PD ✓")

# Scenario 2: quality gate failure — PagerDuty critical
pagerduty_alerts.clear(); slack_alerts.clear()
tiered_alert("critical", "Quality gate FAILED",
             "runs=9 detected in 1 row — delivery_id=9999")
assert len(pagerduty_alerts) == 1
print(f"  Scenario 2 (quality fail): {len(pagerduty_alerts)} PD critical ✓")

# Scenario 3: SLA breach — Slack warning
pagerduty_alerts.clear(); slack_alerts.clear()
sla_breach_monitor = PipelineSLAMonitor()
sla_breach_monitor.start_stage("dbt_build", sla_minutes=0.0001)  # instant breach
time.sleep(0.01)
check = sla_breach_monitor.check_stage("dbt_build")
assert check["breach"] == True
tiered_alert("warning", "SLA breach: dbt_build",
             f"dbt_build took {check['elapsed_min']}min (limit {check['sla_min']}min)")
assert len(slack_alerts) == 1
print(f"  Scenario 3 (SLA breach): {len(slack_alerts)} Slack warning ✓")
print("Step 1 ✓: SLA monitoring and tiered alerting verified")

Step 2 — OpenLineage Emission and Integration Test

Emit OpenLineage START and COMPLETE events for all five pipeline stages with structured dataset references, then run the complete integration test. The integration test asserts: all five stages have emitted events, every event has a non-null job name and at least one input and one output dataset, the warehouse row count is stable (idempotency verified), the dbt mart row count conservation holds, and the alert counts match expectations for a clean run. A full summary is printed showing every assertion's status before the test is marked complete.

Analogy🏏Cricket

🏏 Think of it like cricket: Emitting a START and COMPLETE lineage event for each of the five stages is like the ICC logging, for every phase of the match, exactly which inputs went in and which official record came out — the sensor's scorecard file feeding the extract, the extract feeding the quality gate, and so on down the chain. Each event must carry a named job and at least one input and one output, just as no entry in the audit trail is valid without saying which delivery it came from and which statistic it produced. The integration test is then the technical committee's stumps review: it confirms all five stages logged their events, every event is structurally complete, the warehouse row count held steady (idempotency), the mart conserved its rows, and the clean-run alert counts matched. Running this consolidated pass is what turns five separately-built stages into one defensible claim that the entire pipeline is traceable and correct.

python

# capstone_monitoring.py — Step 2: Lineage emission and integration test
import uuid
from datetime import datetime, timezone

# ── OpenLineage event emitter (best-effort) ────────────────────────────────────
def emit_lineage(
    job_name:   str,
    namespace:  str,
    event_type: str,
    inputs:     list[dict],
    outputs:    list[dict],
    run_id:     str = None,
) -> str:
    run_id = run_id or str(uuid.uuid4())
    event  = {
        "eventType": event_type,
        "eventTime": datetime.now(timezone.utc).isoformat(),
        "run":       {"runId": run_id},
        "job":       {"namespace": namespace, "name": job_name},
        "inputs":    inputs,
        "outputs":   outputs,
    }
    try:
        lineage_log.append(event)
        # In production: requests.post(MARQUEZ_URL, json=event)
    except Exception as exc:
        print(f"  Lineage failed (non-blocking): {exc}")
    return run_id

def ds(namespace: str, name: str) -> dict:
    return {"namespace": namespace, "name": name}

# ── Emit lineage for all five pipeline stages ─────────────────────────────────
lineage_log.clear()

stages = [
    ("ipl.wait_sensor",
     [ds("s3", "ipl-raw/scorecards/2024-04-20/")],
     [ds("airflow", "sensor_pass_token")]),
    ("ipl.extract_matches",
     [ds("ipl-api",   "matches")],
     [ds("postgres",  "staging.deliveries")]),
    ("ipl.quality_gate",
     [ds("postgres",  "staging.deliveries")],
     [ds("postgres",  "warehouse.validated_deliveries")]),
    ("ipl.dbt_build",
     [ds("postgres",  "warehouse.validated_deliveries")],
     [ds("postgres",  "analytics.fct_ipl_bowler_season_stats")]),
    ("ipl.monitor_alert",
     [ds("postgres",  "analytics.fct_ipl_bowler_season_stats")],
     [ds("slack",     "#ipl-data-alerts")]),
]

for job_name, inputs, outputs in stages:
    rid = emit_lineage(job_name, "ipl-data-platform", "START",  inputs, outputs)
    emit_lineage(job_name, "ipl-data-platform", "COMPLETE", inputs, outputs, run_id=rid)

print(f"  Lineage events emitted: {len(lineage_log)} ({len(stages)*2} expected)")
assert len(lineage_log) == len(stages) * 2

# ── Full integration test ────────────────────────────────────────────────────
print("\n=" * 55)
print("Capstone Integration Test — Course 4")
print("=" * 55)

assertions = []

# A1: All pipeline stages have lineage events
jobs_with_events = {e["job"]["name"] for e in lineage_log}
for job_name, _, _ in stages:
    ok = job_name in jobs_with_events
    assertions.append((f"Lineage: {job_name}", ok))

# A2: All lineage events have non-null job and ≥1 input and output
for event in lineage_log:
    ok = (event["job"]["name"] is not None and
          len(event["inputs"]) >= 1 and
          len(event["outputs"]) >= 1)
    assertions.append((f"Event structure: {event['job']['name']}/{event['eventType']}", ok))

# A3: Warehouse idempotency
wh_rows = len(warehouse_db.get("2024-04-20", []))
assertions.append(("Warehouse idempotency (360 rows)", wh_rows == 360))

# A4: dbt mart row conservation
mart_total = int(fct_after["deliveries"].sum())
all_rows   = len(all_del)
assertions.append((f"dbt row conservation ({mart_total}=={all_rows})",
                   mart_total == all_rows))

# A5: Alerts on clean run (0 PD, 1 Slack info)
pagerduty_alerts.clear(); slack_alerts.clear()
send_run_summary("2024-04-20", 3, wh_rows, 0, True, False)
assertions.append(("Clean run: 0 PD alerts",   len(pagerduty_alerts) == 0))
assertions.append(("Clean run: 1 Slack summary",len(slack_alerts) == 1))

# Print results
all_pass = True
for name, ok in assertions:
    icon = "✓" if ok else "✗"
    print(f"  {icon} {name}")
    if not ok: all_pass = False

print()
if all_pass:
    print("ALL ASSERTIONS PASSED — Capstone pipeline is submission-ready ✓")
else:
    failed = [n for n, ok in assertions if not ok]
    raise AssertionError(f"FAILED: {failed}")

Warning: The integration test asserts `mart_total == all_rows`, which requires the incremental simulation to have correctly merged all deliveries including the new batch. If the incremental watermark is set incorrectly — for example, using today's date instead of the epoch sentinel — the new deliveries may be excluded and `mart_total` will be less than `all_rows`, causing the conservation assertion to fail. Always initialise the incremental watermark with a past sentinel value (e.g. `1900-01-01`) to ensure the first run processes all available data, and use `COALESCE(MAX(loaded_at), '1900-01-01')` in the `is_incremental()` condition.

Extension Challenge: Add a lineage validation function that reads the `lineage_log` list and produces a simple lineage graph: for each dataset, which jobs produce it (outputs) and which jobs consume it (inputs). Print the graph in a readable format and verify that the `analytics.fct_ipl_bowler_season_stats` dataset is produced by `ipl.dbt_build` and consumed by `ipl.monitor_alert`. This simulates the Marquez graph traversal that data engineers use for impact analysis in production.