Knowing your pipeline is broken before your stakeholders do.
Data Engineering Module 5 Productionevents table — a standalone checker that exits 0 (OK), 1 (BREACH), or 2 (UNKNOWN), wired as the final task in gharchive_hourly.A pipeline you cannot observe is a pipeline you cannot operate. Before you tune query costs, evolve schemas, or add GDPR controls, you need to answer three questions in under two minutes at 3 am:
| Pillar | What it measures | Canonical signal |
|---|---|---|
| Freshness | How recent is the data? | now − max(created_at) |
| Volume | Are the expected rows arriving? | Row count delta vs trailing average |
| Schema | Has the shape changed unexpectedly? | Column-set hash between runs |
| Distribution | Are values in expected ranges? | Null rate or top-N value share on a key column |
This lesson covers freshness end-to-end. Volume is the Task 1 homework. Schema and distribution land in Lesson 13.
OpenLineage is an open specification (openlineage.io) for emitting lineage events as JSON from any data tool. Every event has the same envelope:
{
"eventType": "START",
"eventTime": "2026-05-26T10:00:00Z",
"producer": "https://github.com/.../data-engineering",
"job": {"namespace": "gharchive-pipeline", "name": "gharchive_hourly.extract"},
"run": {"runId": "4b3a2c1d-..."},
"inputs": [{"namespace": "gharchive-pipeline", "name": "raw_events"}],
"outputs": [{"namespace": "gharchive-pipeline", "name": "dim_events"}]
}Producer → Marquez API (port 5000) → Marquez UI (port 3000)
→ file:// fallback (no Docker needed)
from observability.openlineage_emit import new_run_id, emit_start, emit_complete
def _ol_emit_start(context):
task_id = context["task_instance"].task_id
run_id = new_run_id()
context["task_instance"].xcom_push(key=f"_ol_run_id_{task_id}", value=run_id)
emit_start(run_id, f"gharchive_hourly.{task_id}")
def _ol_emit_complete(context):
task_id = context["task_instance"].task_id
run_id = context["task_instance"].xcom_pull(key=f"_ol_run_id_{task_id}")
if run_id:
emit_complete(run_id, f"gharchive_hourly.{task_id}")
extract = PythonOperator(
task_id="extract",
python_callable=_extract,
pre_execute=_ol_emit_start,
post_execute=_ol_emit_complete,
)Both hooks are wrapped in try/except — an OL backend outage never fails the pipeline.
The rule: agree on the SLA before the alert fires the first time. “Near real-time” is not an SLA. A number in minutes is.
For the gharchive pipeline:
@hourly — up to 60 minutes between runs.30 + 60 + 5 + 25 = 120 minutes worst case → SLA = 90 minutes (target). If age exceeds 90 minutes something upstream is broken.
from observability.freshness import check
# Returns 0 (OK), 1 (BREACH), or 2 (UNKNOWN)
rc = check(db="./warehouse.duckdb", table="events", sla_minutes=90)
Run standalone from the terminal:
cd data-engineering/sample-pipeline
# Check with the real SLA
python -m observability.freshness --sla-minutes 90 ; echo "exit=$?"
# Force a breach (SLA of 0 minutes is always violated)
python -m observability.freshness --sla-minutes 0 ; echo "exit=$?"The exit code is the integration contract: 0 means healthy, 1 means fire an alert, 2 means data is missing (investigate before assuming OK).
The alerts.alert() function emits a single structured JSON line on stdout. Every log collector that understands JSON (Datadog, Loki, CloudWatch) can pick it up with no extra config:
{
"ts": "2026-05-26T11:32:00.123456+00:00",
"severity": "ERROR",
"source": "freshness",
"message": "events table SLA breach",
"dag": "gharchive_hourly"
}Plugging in Slack or PagerDuty is a one-function swap: replace the print(json.dumps(payload)) line with an HTTP POST to your webhook URL. The payload shape stays the same.
After docker compose -f observability/marquez-docker-compose.yml up -d, open http://localhost:3000.
gharchive_hourly.extract).raw_events → dim_events → events. Click any node to see which jobs read or write it.Lineage graph placeholder: after running the DAG three times, the Datasets panel should show three connected nodes. Take a screenshot and paste it here.
now − max(created_at) metric for every table with a downstream SLO. Red when the number exceeds the agreed threshold.An alert requires a human decision. If no human needs to act within 30 minutes, it is not an alert — it is a dashboard metric.
Dashboards are forensics. You look at them after an alert fires to understand what happened, not to discover that something happened.
sample-pipeline/
+ observability/
+ __init__.py
+ openlineage_emit.py # START / COMPLETE event emitter
+ freshness.py # SLA checker; exit 0/1/2
+ alerts.py # JSON line on stdout
+ marquez-docker-compose.yml
airflow/
dags/
- gharchive_hourly.py # extract >> transform
+ gharchive_hourly.py # extract >> transform >> freshness_check
+ # pre_execute/post_execute OpenLineage hooksThe DAG’s original extract and transform tasks are unchanged. The actor column and all upstream schema contracts from Modules 1–4 are preserved. The only additions are the OL hooks and the final freshness_check task.