Homework 12 — Observability & SLAs

Task 1 — Add a volume SLA

Create data-engineering/sample-pipeline/observability/volume.py. This module should check whether the latest hour’s row count in the events table is at least 50% of the trailing 24-hour average. If it falls below that threshold, the pipeline has likely lost data silently.

Requirements:

• The main function should be check(db, table, floor_ratio) returning an exit code: 0 (OK), 1 (BREACH), 2 (UNKNOWN).
• Use DuckDB to query the table; the same database file used by freshness.py.
• Group rows by hour of created_at; compare the latest hour’s count to the average of the preceding 24 hours.
• Print a line in the same style as freshness.py: VOLUME OK: ratio=0.97, VOLUME BREACH: ratio=0.08, or VOLUME UNKNOWN: not enough hours.
• The module must python -m py_compile cleanly.

Self-check: running on the sample data returns VOLUME OK (or VOLUME UNKNOWN if the table has fewer than 2 hours of data). Manually truncating the latest partition to 1 row and re-running returns VOLUME BREACH with the ratio printed.

cd data-engineering/sample-pipeline
python -m py_compile observability/volume.py && echo "syntax OK"
python -m observability.volume --sla-ratio 0.5 ; echo "exit=$?"

Task 2 — Wire the freshness alert into the DAG

Modify the _freshness_check function in data-engineering/sample-pipeline/airflow/dags/gharchive_hourly.py so that a BREACH calls alerts.alert("ERROR", "freshness", ...) before raising RuntimeError.

Requirements:

• The alert JSON line must appear on stdout so any log-collector can capture it.
• The RuntimeError must still be raised after the alert, so Airflow marks the task as failed (not just as having logged an error). A task that logs but does not raise is invisible to downstream alerting.
• The modified DAG must still python -m py_compile cleanly.

Self-check: run a forced breach from the terminal and confirm that a single JSON line with "severity": "ERROR" appears on stdout, followed by the RuntimeError:

cd data-engineering/sample-pipeline
python -c "
import sys
sys.path.insert(0, '.')
# Simulate a freshness check breach inline
from observability import freshness, alerts
rc = freshness.check(sla_minutes=0)
if rc == 1:
    alerts.alert('ERROR', 'freshness', 'events table SLA breach', db='warehouse.duckdb')
    raise RuntimeError('breach')
" 2>&1 | head -5

Task 3 — Read three runs in Marquez (or the JSON event files)

Trigger the gharchive DAG three times (or simulate three runs by running emit_start / emit_complete three times with different run_ids), then verify the results.

from observability.openlineage_emit import new_run_id, emit_start, emit_complete

for _ 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")

After running the code above, verify that exactly six event files were written (three START and three COMPLETE):

ls data-engineering/sample-pipeline/observability/events/ | wc -l

If you have Marquez running, open the Runs panel for gharchive_hourly.extract and confirm the three runs appear on the timeline.

Written reflection (one paragraph, no code required): what would a zero-run gap in the Marquez timeline tell you about your pipeline? Specifically: distinguish between (a) the pipeline tasks failing (which would show failed runs) and (b) the scheduler itself dying (which would show no runs at all). What does the absence of START events tell you that the presence of failed runs does not?

Self-check: six JSON files exist under observability/events/ (plus any from earlier steps); the written reflection addresses the gap-vs-failure distinction in at least two sentences.