Solutions — Homework 12

Lesson 12 · Data Engineering with Python · Observability & SLAs

Task 1 — Volume SLA (volume.py)

"""Volume SLA checker for the gharchive sample pipeline.

Compares the latest hour's row count to the trailing 24-hour average.
Fails the check if the ratio drops below floor_ratio (default 0.5).

Exit codes:
  0  VOLUME OK
  1  VOLUME BREACH
  2  VOLUME UNKNOWN (fewer than 2 hours of data)
"""

from __future__ import annotations

import argparse
import sys

try:
    import duckdb
except ImportError:
    print("volume.py requires duckdb; pip install duckdb", file=sys.stderr)
    sys.exit(2)

DEFAULT_DB = "./warehouse.duckdb"
DEFAULT_TABLE = "events"
DEFAULT_FLOOR = 0.5


def check(
    db: str = DEFAULT_DB,
    table: str = DEFAULT_TABLE,
    floor_ratio: float = DEFAULT_FLOOR,
) -> int:
    """Run the volume check and return an exit code (0/1/2)."""
    con = duckdb.connect(db, read_only=True)
    rows = con.execute(f"""
        SELECT date_trunc('hour', created_at) AS h, COUNT(*) AS n
        FROM {table}
        WHERE created_at >= now() - INTERVAL '25 hours'
        GROUP BY 1
        ORDER BY 1 DESC
    """).fetchall()
    con.close()

    if len(rows) < 2:
        print("VOLUME UNKNOWN: not enough hours of data for comparison.")
        return 2

    latest = rows[0][1]
    avg = sum(n for _, n in rows[1:]) / max(1, len(rows) - 1)
    ratio = latest / avg if avg else 0.0

    if ratio < floor_ratio:
        print(f"VOLUME BREACH: ratio={ratio:.2f} (latest={latest}, avg={avg:.1f})")
        return 1
    print(f"VOLUME OK: ratio={ratio:.2f} (latest={latest}, avg={avg:.1f})")
    return 0


def main() -> None:
    p = argparse.ArgumentParser(description="Check volume SLA for an events table.")
    p.add_argument("--db", default=DEFAULT_DB)
    p.add_argument("--table", default=DEFAULT_TABLE)
    p.add_argument("--sla-ratio", type=float, default=DEFAULT_FLOOR,
                   dest="floor_ratio", help="Minimum ratio of latest/average (0-1)")
    args = p.parse_args()
    sys.exit(check(args.db, args.table, args.floor_ratio))


if __name__ == "__main__":
    main()

Common mistake: using a fixed row-count floor instead of a ratio. gharchive volume varies roughly 3× across the day (more events during US/EU business hours than at night). A floor of 5000 rows would breach every night even when the pipeline is healthy. A ratio of 50% of the 24-hour average adapts to natural daily volume patterns and only fires when something genuinely unusual has happened.

Task 2 — Wire alerts into the DAG

The relevant section of gharchive_hourly.py after the edit (only the _freshness_check function body changes):

def _freshness_check(**_context) -> None:
    """Run the observability freshness SLA check; raise on breach.

    L12 (updated): a BREACH calls alerts.alert() before raising so that
    any log collector captures a structured JSON event, not just a Python
    traceback.
    """
    try:
        from observability.freshness import check
        from observability.alerts import alert
    except ImportError as exc:
        print(f"freshness_check: observability package unavailable ({exc}), skipping.")
        return

    rc = check(sla_minutes=90)
    if rc == 1:
        alert("ERROR", "freshness", "events table SLA breach",
              dag="gharchive_hourly", sla_minutes=90)
        raise RuntimeError("Freshness SLA breach: events table age exceeds 90 minutes.")
    if rc == 2:
        alert("WARN", "freshness", "events table UNKNOWN — may be empty",
              dag="gharchive_hourly")

Common mistake: swallowing the breach by only logging it. A task that calls alerts.alert() but does not raise will show green in Airflow. The downstream on-call rotation, which watches for red tasks, never fires. The alert JSON line goes into the logs but not into the on-call channel. Always raise after alerting — the DAG failure is the integration with PagerDuty/OpsGenie that turns the structured log line into a page.

Task 3 — Three runs in Marquez / JSON files

Worked answer (code)

from observability.openlineage_emit import new_run_id, emit_start, emit_complete

for i in range(3):
    rid = new_run_id()
    emit_start(rid, "gharchive_hourly.extract",
               inputs=["raw_events"], outputs=["dim_events"])
    emit_complete(rid, "gharchive_hourly.extract")
    print(f"run {i+1}: {rid}")

# Verify: should print 6 (or more if earlier steps added files)
ls data-engineering/sample-pipeline/observability/events/ | wc -l

Reflection: zero-run gap vs failed runs

A zero-run gap in the Marquez timeline — a period with no START events at all — is a different failure signal than a run that failed. A failed run means the scheduler attempted the task and the task raised an exception: the job tried, the job broke, and the job told you. A gap means the scheduler never attempted the task at all. No START event was emitted because no process ran to emit one. This is typically a scheduler outage (the Airflow scheduler died, ran out of memory, or lost its database connection), not a pipeline failure. It is also the harder failure mode to detect, because a dead scheduler produces no logs, no red tasks, and no alerts from the pipeline itself.

This is why Marquez’s run-timeline view is the first dashboard to check at 3 am. The absence of START events is a stronger signal than the presence of failed runs: it tells you that the entire orchestration layer, not just one pipeline task, has stopped functioning. When you see the gap, the first question is “is the Airflow scheduler running?” — not “which task failed?”. Lesson 13 turns this lineage from a per-pipeline view into a full end-to-end graph so that a gap in any one pipeline is visible against all pipelines at once.