Lab 12 — Instrument the Sample Pipeline with OpenLineage & Freshness SLA

Lesson 12 · Data Engineering with Python · ~40 min

By the end of this lab every DAG task emits OpenLineage events, the freshness check exit code is visible in your terminal, and one synthetic SLA breach fires the alerts.alert() line. No code changes are required — the observability/ tree was added to the sample-pipeline in the L12 commit, and the DAG was updated to use it. This lab walks you through verifying the wiring.

Prerequisites

• Sample-pipeline virtual environment from Plans A–D active. If you have not set it up: cd data-engineering/sample-pipeline && python -m venv .venv && source .venv/bin/activate
• Python 3.10+.
• duckdb installed (pip install duckdb) — required by freshness.py.
• openlineage-python is optional. The file-backend fallback works without it.
• Docker Desktop running if you want the Marquez UI (Step 5). Skippable.

Step 1 — Install or verify observability deps

cd data-engineering/sample-pipeline
pip install -r requirements.txt

Plan E added openlineage-python, openlineage-airflow, and duckdb to requirements.txt. If the install fails on your machine (e.g., C-extension compilation error on openlineage), skip it — the file-backend fallback in openlineage_emit.py engages automatically and writes JSON files locally instead of posting to Marquez.

Step 2 — Inspect the new observability/ tree

ls data-engineering/sample-pipeline/observability/

You should see:

__init__.py
openlineage_emit.py
freshness.py
alerts.py
marquez-docker-compose.yml

Read the inline docstrings; nothing needs changing. Key points to understand:

• openlineage_emit.py — emit_start(run_id, job_name) and emit_complete(run_id, job_name). Uses OPENLINEAGE_URL env var; defaults to file:// if unset or if openlineage-python is not installed.
• freshness.py — reads MAX(created_at) from a DuckDB table; computes age; returns exit code 0/1/2. Callable as a module or from Python.
• alerts.py — one function: alert(severity, source, message). Emits a JSON line to stdout.

Step 3 — Run the smoke test

cd data-engineering/sample-pipeline

# Fire a test alert
python -m observability.alerts WARN demo "hello from alerts"

# Emit two OpenLineage events (START + COMPLETE) via the file fallback
python -c "
from observability.openlineage_emit import new_run_id, emit_start, emit_complete
rid = new_run_id()
emit_start(rid, 'smoke', inputs=['raw'], outputs=['dim'])
emit_complete(rid, 'smoke')
"

# Inspect the JSON files that were written
ls observability/events/

Expected:

• One JSON line printed (the alert): {"ts": "...", "severity": "WARN", "source": "demo", "message": "hello from alerts"}
• Two JSON files under observability/events/: one ending in -START.json, one ending in -COMPLETE.json.

Step 4 — Run a DAG end-to-end with lineage

Start Airflow (the docker-compose.yml from L04 in airflow/):

cd data-engineering/sample-pipeline
docker compose -f airflow/docker-compose.yml up airflow-webserver airflow-scheduler -d

Trigger the gharchive_hourly DAG from the Airflow UI or CLI:

docker compose -f airflow/docker-compose.yml exec airflow-scheduler \
  airflow dags trigger gharchive_hourly

Watch the task logs. Each of the three tasks (extract, transform, freshness_check) will log emit_start and emit_complete calls. The freshness_check task at the end of the DAG should print:

FRESHNESS OK: latest=... age=... sla=0:01:30

Step 5 — Bring up Marquez (optional)

cd data-engineering/sample-pipeline
docker compose -f observability/marquez-docker-compose.yml up -d
export OPENLINEAGE_URL=http://localhost:5000

Re-trigger the DAG (or re-run the smoke test from Step 3 with OPENLINEAGE_URL set). Open http://localhost:3000. Navigate to:

• Datasets → you should see raw, dim, and events nodes connected by the job that wrote and read them.
• Jobs → gharchive_hourly.smoke → Runs — a timeline with one completed run.

Step 6 — Check freshness

cd data-engineering/sample-pipeline
python -m observability.freshness --sla-minutes 90 ; echo "exit=$?"

Expected output:

FRESHNESS OK: latest=... age=... sla=0:01:30
exit=0

If the events table does not exist (you have not run the L10 Iceberg write step), the exit code will be 2 (UNKNOWN). That is expected — see the Troubleshooting section.

Step 7 — Trigger a synthetic SLA breach

Force a breach by setting the SLA to zero minutes (any positive age will violate it):

python -m observability.freshness --sla-minutes 0 ; echo "exit=$?"

Expected output:

FRESHNESS BREACH: latest=... age=... sla=0:00:00
exit=1

Now pipe the breach exit code to the alert helper:

python -m observability.freshness --sla-minutes 0 \
  || python -m observability.alerts ERROR freshness "events table SLA breach"

Expected: the freshness breach line is printed first (to stdout), then the alert JSON line: {"ts": "...", "severity": "ERROR", "source": "freshness", "message": "events table SLA breach"}

What you should leave with

Troubleshooting