Skip to main content

Pipeline Observability

A green task says the process returned success. It does not prove the expected partition arrived, every row validated, duplicates were controlled, totals reconciled, or consumers received fresh data. Reliable pipelines observe both the control plane—runs, attempts, durations, queues—and the data plane—volume, schema, keys, nulls, rejects, freshness, and business invariants.

Prerequisites​

Complete Airflow DAG Fundamentals. This final lesson measures the same event pipeline, including duplicate, late, and malformed deliveries.

Learning goals​

  • Emit metrics with stable meaning and bounded cardinality.
  • Connect logs and data-quality evidence with run and partition identity.
  • Design symptom-based alerts with owner, threshold, and runbook.
  • Distinguish freshness, completeness, validity, uniqueness, and correctness.

Worked example​

The fixed observation time makes freshness deterministic. Invalid numeric input is rejected explicitly; duplicate and late deliveries remain visible instead of disappearing inside a “success” count.

import json
from datetime import datetime
from decimal import Decimal, InvalidOperation
from pathlib import Path

def instant(value):
    return datetime.fromisoformat(value.replace("Z", "+00:00"))

root = Path("fixtures/data-engineering/pipelines")
deliveries = []
for name in ("events-batch-1.jsonl", "events-batch-2.jsonl", "events-late.jsonl"):
    deliveries.extend(json.loads(line) for line in (root / name).read_text().splitlines())
malformed = [json.loads(line) for line in (root / "events-malformed.jsonl").read_text().splitlines()]

valid = {}
for event in deliveries:
    valid[event["event_id"]] = event
rejected = 0
for event in malformed:
    try:
        Decimal(event["amount"])
    except InvalidOperation:
        rejected += 1

max_on_time_event = max(instant(event["occurred_at"]) for event in valid.values()
                        if event["arrived_at"] < "2026-10-03")
late_events = sum(instant(event["occurred_at"]) < max_on_time_event
                  for event in valid.values() if event["arrived_at"] >= "2026-10-03")
now = instant("2026-10-03T13:00:00Z")
latest_arrival = max(instant(event["arrived_at"]) for event in valid.values())
metrics = {
    "delivered_records": len(deliveries),
    "duplicate_deliveries": len(deliveries) - len(valid),
    "freshness_seconds": int((now - latest_arrival).total_seconds()),
    "late_events": late_events,
    "published_rows": len(valid),
    "rejected_records": rejected,
    "total_amount_cents": sum(int(Decimal(e["amount"]) * 100) for e in valid.values()),
}
print(json.dumps(metrics, sort_keys=True))

The metrics report seven deliveries, one duplicate, six published rows, one late event, one rejected record, 4,550 cents, and 3,600 seconds of freshness lag. These values are assertions for the fixture, not universal alert thresholds.

Attach every log and metric to stable dimensions such as pipeline, task, run, environment, and partition. Do not use unbounded event_id values as metric labels; keep record-level details in sampled or access-controlled logs and quarantine artifacts. Trace lineage from input snapshot and code version to output partition so an operator can answer what changed and what to replay.

Alerts should describe user-visible symptoms: a partition is past its freshness objective, reject rate exceeds an agreed bound, reconciliation fails, or retries are exhausted. Every alert needs an owner, urgency, dashboard, diagnostic query, mitigation, and recovery verification. Alerting on every retry creates noise; the final failed state or threatened service objective is usually more actionable.

Limitations and common mistakes​

  • A dashboard without explicit units, windows, and denominators is ambiguous.
  • Row-count equality cannot detect wrong values or duplicated-and-missing pairs.
  • High-cardinality labels can make metric systems expensive or unusable.
  • Logs may contain personal or secret data; apply minimization, access, and retention controls.
  • Observability detects symptoms; it does not replace contracts, idempotence, backups, or drills.

Practice​

Define one service-level indicator and objective for daily publication. Add an alert condition, owner, first diagnostic query, safe mitigation, and the evidence required to close the incident.

Next steps​

You can now design a locally repeatable pipeline from ingestion through recovery and observation. Continue with the next Data Engineering batch on analytics engineering, beginning with dimensional modeling. Revisit ETL vs. ELT when a new source changes trust boundaries.

Primary sources​

Checked 2026-10-15: OpenTelemetry metrics, OpenTelemetry logs, Airflow metrics, and Airflow logging.