A report is wrong. Which of the forty upstream tables changed? Which pipeline ran late? Did the source deliver? Who else is affected? Data observability is the practice of being able to answer those questions in minutes from telemetry, instead of in hours from a chain of Slack messages. It combines the quality checks of an earlier module with pipeline metrics, freshness monitoring, and lineage — the graph of which table feeds which — into a picture of the platform's health that both engineers and consumers can read. This module covers the signals to collect, how to build and use lineage, SLAs for data products, incident handling for data, and the dashboard that tells consumers whether to trust what they see.
- Define the signals of data observability: freshness, volume, schema, distribution, lineage, pipeline run metrics
- Collect pipeline telemetry (runs, durations, rows, failures) and expose it as metrics and logs
- Build lineage from the transformation layer and orchestrator, and use it for impact and root-cause analysis
- Set data SLAs with owners and measure them
- Run a data incident: detect, scope with lineage, communicate, fix, and learn
The signals
Observability for software asks whether the service is up and fast; for data it asks whether the tables are current, complete and correct, and whether the pipelines that produce them ran. The signals map onto the quality dimensions plus the pipeline itself: freshness (when did each table last update, versus when it should have), volume (rows per run versus baseline), schema (changes over time), distribution (key metrics drifting), lineage (dependencies, so impact can be traced), and pipeline health (run status, duration, retries, cost). The first four are properties of tables; the last two are properties of the system that makes them.
| Signal | Source | Where it surfaces |
|---|---|---|
| Freshness | max(updated_at) per table; run completion times | Per-table SLA dashboard; alert on breach |
| Volume and distribution | Row counts and metric snapshots per run | Anomaly alerts with baselines |
| Schema | Warehouse information schema diffs; contract checks | Change log; alert on breaking change |
| Lineage | dbt manifest; orchestrator graph; query logs | Graph UI; impact and root-cause queries |
| Pipeline runs | Orchestrator metadata; job logs | Run metrics, duration trends, failure alerts |
Observability tools (open-source and commercial) package all of this. The concepts matter more than the tool: whatever you use, you need the same signals, the same lineage, and the same ownership.
Pipeline telemetry
Every pipeline run should leave a record: pipeline and task name, logical date, start and end time, status, rows in and out, rows rejected, bytes processed, and cost where the platform exposes it. Orchestrators store most of this already; the gaps are the row counts and business-level numbers, which the pipeline must emit itself. Publish them as metrics (a Prometheus gauge or counter per pipeline and table, or a metrics table in the warehouse) so they can be graphed and alerted on with the same tooling the SRE track uses for services.
import time
from datetime import date, datetime, timezone
from prometheus_client import Counter, Gauge, Histogram, push_to_gateway, CollectorRegistry
registry = CollectorRegistry()
rows_out = Gauge("pipeline_rows_out", "Rows written by the last run", ["pipeline", "table"], registry=registry)
run_seconds = Histogram("pipeline_run_seconds", "Run duration", ["pipeline"], registry=registry)
last_success = Gauge("pipeline_last_success_timestamp", "Unix time of the last successful run", ["pipeline"], registry=registry)
failures = Counter("pipeline_failures_total", "Failed runs", ["pipeline"], registry=registry)
def record_run(conn, pipeline: str, table: str, logical_date: date, started: float, n_out: int, n_rejected: int, ok: bool):
finished = time.time()
conn.execute(
"INSERT INTO ops.pipeline_runs (pipeline, table_name, logical_date, started_at, finished_at, status, rows_out, rows_rejected) "
"VALUES (%s, %s, %s, to_timestamp(%s), to_timestamp(%s), %s, %s, %s)",
(pipeline, table, logical_date, started, finished, "success" if ok else "failed", n_out, n_rejected),
)
rows_out.labels(pipeline, table).set(n_out)
run_seconds.labels(pipeline).observe(finished - started)
if ok:
last_success.labels(pipeline).set(finished)
else:
failures.labels(pipeline).inc()
push_to_gateway("pushgateway.internal:9091", job=pipeline, registry=registry)With pipeline_last_success_timestamp in Prometheus, freshness is an alert rule: time() - pipeline_last_success_timestamp{pipeline="orders_daily"} > 6 * 3600. With ops.pipeline_runs in the warehouse, the same question is a query, and volume baselines are a window function over the last four same-weekday runs. Duration trends catch the pipeline that is slowly getting slower before it misses its window.
Lineage: the map
Lineage is the graph of dependencies: source system to raw table to staging model to mart to dashboard. It answers the two questions every data incident starts with. Impact: if this table is wrong or late, what is downstream of it and who owns those things? Root cause: if this dashboard is wrong, what is upstream of it, and which of those changed or failed recently? Column-level lineage extends this to individual fields, which is what you need to answer "where does this number come from" and "what breaks if I drop this column".
Lineage is mostly free if the platform is built as earlier modules described: dbt's manifest holds every model's refs and sources; the orchestrator holds task dependencies and which runs produced which tables; warehouse query logs can be parsed for table-to-table flows outside dbt; BI tools expose which dashboards read which tables. The OpenLineage standard lets orchestrators and processing engines emit run and dataset events in one format, and catalogs (open-source ones like DataHub or OpenMetadata, or commercial) assemble them into one navigable graph with owners attached.
import json
from collections import deque
manifest = json.load(open("target/manifest.json"))
children = manifest["child_map"] # node id -> list of dependent node ids
nodes = {**manifest["nodes"], **manifest.get("exposures", {})}
def downstream(node_id: str) -> list[tuple[str, str]]:
seen, queue, out = set(), deque([node_id]), []
while queue:
cur = queue.popleft()
for child in children.get(cur, []):
if child not in seen:
seen.add(child)
queue.append(child)
meta = nodes.get(child, {})
out.append((child, meta.get("config", {}).get("meta", {}).get("owner", meta.get("owner", "?"))))
return out
for node, owner in downstream("model.shop.stg_shop__orders"):
print(f"{node:<55} owner={owner}")
# includes exposures: the dashboards declared in dbt that read fct_ordersDeclare dashboards and downstream applications as dbt exposures. They then appear in lineage with an owner, and "which dashboards break if this model fails" becomes a query instead of a guess.
SLAs for data products
A table that people depend on is a product, and products have service levels. A data SLA states, per product: freshness (updated by 06:00 UTC daily), completeness (row count within tolerance, no missing partitions), quality (the checks that must pass), and support (an owner, a channel, a response time). Set them with the consumers, measure them with the telemetry above, and report compliance monthly. As in the SRE track, the point is not the number itself but the conversation it forces: what does this dashboard actually need, and what does it cost to guarantee it.
models:
- name: fct_orders
description: One row per non-cancelled order. Revenue source of truth for finance dashboards.
meta:
owner: data-platform@acme.example
product_tier: 1
sla:
freshness: "updated by 06:00 UTC every day"
completeness: "daily row count within 30% of same-weekday baseline; no missing partitions in trailing 7 days"
quality: "all tests pass; unique order_id; no negative amounts"
support: "#data-platform, response within 1 business hour for tier 1"
tests:
- dbt_utils.recency: {datepart: hour, field: loaded_at, interval: 8}Tier the products. Tier 1 (finance, executive, customer-facing) gets strict SLAs, paging alerts and the most checks; tier 3 (exploratory, internal) gets freshness monitoring and a best-effort owner. Trying to guarantee everything guarantees nothing; tiering concentrates the effort where a failure costs the most.
Data incidents and the trust dashboard
A data incident follows the SRE incident shape with two data-specific steps. Detect: a freshness, volume or quality alert, or a consumer report. Scope with lineage: walk upstream to find the failing or changed node, walk downstream to list every affected product and owner. Communicate: post in the product's channel and on the trust dashboard which tables are affected, since when, and the expected fix time; consumers who know the numbers are stale will wait instead of building spreadsheets. Fix and backfill with the idempotent tools from earlier modules. Learn: record time to detect and time to fix, and add the check or the lineage declaration that would have shortened either.
- A trust dashboard per product or per domain: last update time, SLA status, checks passed or failed, open incidents. Embed a short version on the BI dashboard itself.
- Runbooks per pipeline: how to rerun, how to backfill, who owns the source, what usually breaks.
- Incident log with time to detect, time to resolve and affected products; review monthly for patterns.
- Change notifications: a schema change or a model change to a tier-1 product announced before it merges, with lineage listing who to tell.
The most damaging data incidents are the quiet ones: a stale table that still looks plausible. Freshness on every product, visible to consumers, is the cheapest defence.