Most pipelines tell you whether they ran — not whether they did the right thing. This lesson walks you through designing and building a production-grade metadata store in Python and PostgreSQL that captures run history, tracks data volumes, and maintains a full audit trail, all queryable in SQL.
Picture this: it's 9 AM on a Monday, your sales team is complaining that last week's revenue report looks wrong, and your manager is asking whether the pipeline ran successfully over the weekend. You open your orchestration tool, see a green checkmark, and think you're in the clear — until you realize the pipeline ran but processed zero rows due to an upstream schema change. The job "succeeded" by every technical measure. Nothing failed. Nobody was alerted. And now you're manually digging through logs trying to reconstruct what happened three days ago.
This scenario plays out constantly in production data engineering. Orchestration tools like Airflow tell you whether a task ran. They don't tell you how many rows it processed, whether those numbers were in a reasonable range, how long each stage took compared to last week, or which specific records were modified during a load. That gap between "the pipeline ran" and "the pipeline did the right thing" is exactly where a pipeline metadata store lives.
By the end of this lesson, you'll have a working metadata store implementation that captures run history, tracks data volumes, and maintains a full audit trail — all queryable in SQL so you can answer operational questions without grepping through log files.
What you'll learn:
You should be comfortable writing Python data pipelines and working with SQL. Familiarity with building your first data pipeline with Python will help, as will a basic understanding of logging, alerting, and observability for data pipelines — we'll extend those concepts here into structured, queryable storage rather than log files.
You'll need Python 3.9+, SQLAlchemy 2.x, and access to a PostgreSQL instance (or SQLite for local development).
Before we write any code, it's worth being precise about what we're building and why it's distinct from your existing logging setup.
Log files (and log aggregation tools like Datadog or CloudWatch) are optimized for searching across unstructured text. They're great for debugging individual failures. But they're terrible for answering questions like "What's the average row count for the customer_orders pipeline over the last 30 days?" or "Show me every pipeline run that touched the orders table between January 1st and January 15th."
A metadata store is a structured database — usually a simple relational schema — that your pipelines write to programmatically. Every run produces a record. Every record contains structured fields: timestamps, row counts, status codes, source/target identifiers, durations. Because it's relational, you can aggregate it, join it, and query it with SQL you already know.
Think of it as the difference between your pipeline's journal (logs) and its ledger (metadata store). Both matter. The ledger is what you reach for when you need to answer a business question about pipeline behavior over time.
Key insight: A metadata store doesn't replace your logging infrastructure — it complements it. Logs give you the "what went wrong" narrative; the metadata store gives you the "what has been happening" trend data. You need both in production.
The schema design is where most teams make their first mistake: they create a single pipeline_runs table and stuff everything into it. That works until you have pipelines with multiple stages, and then you're either duplicating rows or concatenating structured data into a text column — both of which make querying painful.
We'll use a three-level hierarchy: pipelines, runs, and tasks.
customer_orders_daily)extract, transform, load), each with its own timing and data volume metricsHere's the full schema:
-- The canonical registry of pipelines
CREATE TABLE pipelines (
pipeline_id SERIAL PRIMARY KEY,
pipeline_name VARCHAR(255) NOT NULL UNIQUE,
pipeline_version VARCHAR(50) NOT NULL DEFAULT '1.0.0',
description TEXT,
owner_team VARCHAR(100),
source_system VARCHAR(100),
target_system VARCHAR(100),
created_at TIMESTAMPTZ NOT NULL DEFAULT NOW(),
updated_at TIMESTAMPTZ NOT NULL DEFAULT NOW()
);
-- One record per execution of a pipeline
CREATE TABLE pipeline_runs (
run_id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
pipeline_id INTEGER NOT NULL REFERENCES pipelines(pipeline_id),
run_status VARCHAR(50) NOT NULL DEFAULT 'RUNNING',
-- Values: RUNNING, SUCCESS, FAILED, PARTIAL, SKIPPED
trigger_type VARCHAR(50),
-- Values: SCHEDULED, MANUAL, EVENT, BACKFILL
triggered_by VARCHAR(255),
environment VARCHAR(50) NOT NULL DEFAULT 'production',
run_start_time TIMESTAMPTZ NOT NULL DEFAULT NOW(),
run_end_time TIMESTAMPTZ,
duration_seconds INTEGER,
rows_read BIGINT DEFAULT 0,
rows_written BIGINT DEFAULT 0,
rows_rejected BIGINT DEFAULT 0,
data_interval_start TIMESTAMPTZ, -- The business period this run covers
data_interval_end TIMESTAMPTZ,
error_message TEXT,
run_metadata JSONB, -- Flexible bag for run-specific context
created_at TIMESTAMPTZ NOT NULL DEFAULT NOW()
);
CREATE INDEX idx_pipeline_runs_pipeline_id ON pipeline_runs(pipeline_id);
CREATE INDEX idx_pipeline_runs_status ON pipeline_runs(run_status);
CREATE INDEX idx_pipeline_runs_start_time ON pipeline_runs(run_start_time DESC);
-- One record per task/stage within a run
CREATE TABLE task_runs (
task_run_id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
run_id UUID NOT NULL REFERENCES pipeline_runs(run_id),
task_name VARCHAR(255) NOT NULL,
task_status VARCHAR(50) NOT NULL DEFAULT 'RUNNING',
start_time TIMESTAMPTZ NOT NULL DEFAULT NOW(),
end_time TIMESTAMPTZ,
duration_seconds INTEGER,
rows_in BIGINT DEFAULT 0,
rows_out BIGINT DEFAULT 0,
rows_rejected BIGINT DEFAULT 0,
bytes_processed BIGINT DEFAULT 0,
error_message TEXT,
task_metadata JSONB
);
CREATE INDEX idx_task_runs_run_id ON task_runs(run_id);
-- Audit trail: one record per meaningful data event
CREATE TABLE audit_events (
event_id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
run_id UUID NOT NULL REFERENCES pipeline_runs(run_id),
task_run_id UUID REFERENCES task_runs(task_run_id),
event_time TIMESTAMPTZ NOT NULL DEFAULT NOW(),
event_type VARCHAR(100) NOT NULL,
-- Values: SCHEMA_CHANGE, RECORD_INSERTED, RECORD_UPDATED,
-- RECORD_DELETED, VALIDATION_FAILURE, THRESHOLD_BREACH
target_table VARCHAR(255),
target_schema VARCHAR(255),
record_count BIGINT,
old_value JSONB, -- For schema changes or record updates
new_value JSONB,
event_metadata JSONB
);
CREATE INDEX idx_audit_events_run_id ON audit_events(run_id);
CREATE INDEX idx_audit_events_target_table ON audit_events(target_table);
CREATE INDEX idx_audit_events_event_time ON audit_events(event_time DESC);
Notice a few deliberate design decisions here:
The run_id is a UUID, not a serial integer. This matters because pipelines often run across distributed systems — you want to be able to generate the run ID in your pipeline code before you write to the database, so you can include it in log messages and correlate them with metadata store records from the very start of a run.
The data_interval_start and data_interval_end fields capture the business period the run covers, not just when it ran. If you're running a pipeline at 2 AM to process yesterday's orders, those fields hold yesterday's date range. This distinction is critical for incremental loading patterns where you need to know exactly what time window a run was responsible for.
The run_metadata and task_metadata JSONB columns give you flexibility without requiring schema migrations every time you want to track a new piece of context. Use them for things like "which S3 prefix was read" or "which dbt model version was used" — structured enough to be useful, flexible enough to not require a migration.
Warning: Don't put everything in JSONB. The fixed columns (
rows_read,rows_written,duration_seconds) should stay as typed columns because you'll aggregate and filter on them constantly. JSONB queries are slower and less ergonomic for high-frequency analytical queries.
Now let's build the Python class that your pipelines will use to write to this store. The goal is a clean interface that makes instrumentation low-friction — if adding metadata tracking requires 50 lines of boilerplate per pipeline, teams will skip it.
# metadata_store/client.py
import uuid
import time
from contextlib import contextmanager
from datetime import datetime, timezone
from typing import Optional, Dict, Any
from sqlalchemy import create_engine, text
from sqlalchemy.orm import Session
class MetadataStore:
"""
Client for writing pipeline execution metadata to a structured store.
Designed to be instantiated once per pipeline run and passed to tasks.
"""
def __init__(self, db_url: str, pipeline_name: str, environment: str = "production"):
self.engine = create_engine(db_url, pool_pre_ping=True)
self.pipeline_name = pipeline_name
self.environment = environment
self._pipeline_id: Optional[int] = None
self._run_id: Optional[str] = None
def _get_pipeline_id(self) -> int:
"""Resolve pipeline name to ID, creating the registry entry if needed."""
with Session(self.engine) as session:
result = session.execute(
text("SELECT pipeline_id FROM pipelines WHERE pipeline_name = :name"),
{"name": self.pipeline_name}
).fetchone()
if result:
return result[0]
# Auto-register unknown pipelines — useful during development,
# but consider making this a warning in strict production environments
result = session.execute(
text("""
INSERT INTO pipelines (pipeline_name)
VALUES (:name)
RETURNING pipeline_id
"""),
{"name": self.pipeline_name}
).fetchone()
session.commit()
return result[0]
def start_run(
self,
trigger_type: str = "SCHEDULED",
triggered_by: str = "system",
data_interval_start: Optional[datetime] = None,
data_interval_end: Optional[datetime] = None,
run_metadata: Optional[Dict[str, Any]] = None,
) -> str:
"""
Register the start of a pipeline run.
Returns the run_id — store this and pass it to all downstream tasks.
"""
self._pipeline_id = self._get_pipeline_id()
self._run_id = str(uuid.uuid4())
with Session(self.engine) as session:
session.execute(
text("""
INSERT INTO pipeline_runs (
run_id, pipeline_id, run_status, trigger_type,
triggered_by, environment, run_start_time,
data_interval_start, data_interval_end, run_metadata
) VALUES (
:run_id, :pipeline_id, 'RUNNING', :trigger_type,
:triggered_by, :environment, :start_time,
:data_interval_start, :data_interval_end, :run_metadata
)
"""),
{
"run_id": self._run_id,
"pipeline_id": self._pipeline_id,
"trigger_type": trigger_type,
"triggered_by": triggered_by,
"environment": self.environment,
"start_time": datetime.now(timezone.utc),
"data_interval_start": data_interval_start,
"data_interval_end": data_interval_end,
"run_metadata": run_metadata,
}
)
session.commit()
return self._run_id
def complete_run(
self,
rows_read: int = 0,
rows_written: int = 0,
rows_rejected: int = 0,
status: str = "SUCCESS",
error_message: Optional[str] = None,
):
"""Mark a run as complete with final volume metrics."""
end_time = datetime.now(timezone.utc)
with Session(self.engine) as session:
session.execute(
text("""
UPDATE pipeline_runs
SET run_status = :status,
run_end_time = :end_time,
duration_seconds = EXTRACT(EPOCH FROM (:end_time - run_start_time))::INTEGER,
rows_read = :rows_read,
rows_written = :rows_written,
rows_rejected = :rows_rejected,
error_message = :error_message
WHERE run_id = :run_id
"""),
{
"status": status,
"end_time": end_time,
"rows_read": rows_read,
"rows_written": rows_written,
"rows_rejected": rows_rejected,
"error_message": error_message,
"run_id": self._run_id,
}
)
session.commit()
def fail_run(self, error_message: str):
"""Convenience method for marking a run as failed."""
self.complete_run(status="FAILED", error_message=error_message)
@contextmanager
def task(self, task_name: str, task_metadata: Optional[Dict[str, Any]] = None):
"""
Context manager for wrapping individual pipeline tasks.
Automatically tracks timing and handles success/failure status.
Usage:
with store.task("extract_orders") as t:
rows = extract_from_source()
t.rows_in = len(rows)
"""
task_run_id = str(uuid.uuid4())
start_time = datetime.now(timezone.utc)
tracker = TaskTracker(task_run_id)
with Session(self.engine) as session:
session.execute(
text("""
INSERT INTO task_runs (task_run_id, run_id, task_name, task_status, start_time, task_metadata)
VALUES (:task_run_id, :run_id, :task_name, 'RUNNING', :start_time, :task_metadata)
"""),
{
"task_run_id": task_run_id,
"run_id": self._run_id,
"task_name": task_name,
"start_time": start_time,
"task_metadata": task_metadata,
}
)
session.commit()
try:
yield tracker
status = "SUCCESS"
error_message = None
except Exception as e:
status = "FAILED"
error_message = str(e)
raise
finally:
end_time = datetime.now(timezone.utc)
duration = int((end_time - start_time).total_seconds())
with Session(self.engine) as session:
session.execute(
text("""
UPDATE task_runs
SET task_status = :status,
end_time = :end_time,
duration_seconds = :duration,
rows_in = :rows_in,
rows_out = :rows_out,
rows_rejected = :rows_rejected,
bytes_processed = :bytes_processed,
error_message = :error_message
WHERE task_run_id = :task_run_id
"""),
{
"status": status,
"end_time": end_time,
"duration": duration,
"rows_in": tracker.rows_in,
"rows_out": tracker.rows_out,
"rows_rejected": tracker.rows_rejected,
"bytes_processed": tracker.bytes_processed,
"error_message": error_message,
"task_run_id": task_run_id,
}
)
session.commit()
def log_audit_event(
self,
event_type: str,
target_table: str,
target_schema: str = "public",
record_count: Optional[int] = None,
old_value: Optional[Dict] = None,
new_value: Optional[Dict] = None,
task_run_id: Optional[str] = None,
event_metadata: Optional[Dict] = None,
):
"""Write a discrete audit event to the audit trail."""
with Session(self.engine) as session:
session.execute(
text("""
INSERT INTO audit_events (
run_id, task_run_id, event_type, target_table,
target_schema, record_count, old_value, new_value, event_metadata
) VALUES (
:run_id, :task_run_id, :event_type, :target_table,
:target_schema, :record_count, :old_value, :new_value, :event_metadata
)
"""),
{
"run_id": self._run_id,
"task_run_id": task_run_id,
"event_type": event_type,
"target_table": target_table,
"target_schema": target_schema,
"record_count": record_count,
"old_value": old_value,
"new_value": new_value,
"event_metadata": event_metadata,
}
)
session.commit()
class TaskTracker:
"""Mutable metrics holder passed into task context managers."""
def __init__(self, task_run_id: str):
self.task_run_id = task_run_id
self.rows_in: int = 0
self.rows_out: int = 0
self.rows_rejected: int = 0
self.bytes_processed: int = 0
The task() context manager is the workhorse here. It handles the start/end database writes automatically, so you don't have to remember to close a task record — even if an exception is raised. The TaskTracker object it yields is a simple mutable holder for metrics that your task code can update as it runs.
Let's see what this looks like when applied to a realistic pipeline — a daily load of customer order data from a transactional PostgreSQL database into a data warehouse.
# pipelines/customer_orders_daily.py
import json
from datetime import datetime, timezone, timedelta
from typing import List, Dict
import pandas as pd
from metadata_store.client import MetadataStore
METADATA_DB_URL = "postgresql://meta_user:password@metadata-db:5432/pipeline_meta"
SOURCE_DB_URL = "postgresql://ro_user:password@orders-db:5432/orders"
WAREHOUSE_URL = "postgresql://dw_user:password@warehouse:5432/analytics"
def run_customer_orders_pipeline(
data_interval_start: datetime,
data_interval_end: datetime,
triggered_by: str = "airflow",
):
store = MetadataStore(
db_url=METADATA_DB_URL,
pipeline_name="customer_orders_daily",
environment="production",
)
run_id = store.start_run(
trigger_type="SCHEDULED",
triggered_by=triggered_by,
data_interval_start=data_interval_start,
data_interval_end=data_interval_end,
run_metadata={
"source_db": "orders-db",
"target_schema": "analytics.fact_orders",
"git_sha": "a3f9b12", # Pull from env in real life
},
)
total_rows_read = 0
total_rows_written = 0
total_rows_rejected = 0
try:
# --- EXTRACT ---
raw_orders: List[Dict] = []
with store.task("extract_orders") as t:
# In real code, use your connection pool here
source_engine = create_engine(SOURCE_DB_URL)
df = pd.read_sql(
"""
SELECT order_id, customer_id, order_date, total_amount,
status, line_item_count, updated_at
FROM orders
WHERE updated_at >= %(start)s AND updated_at < %(end)s
""",
source_engine,
params={"start": data_interval_start, "end": data_interval_end},
)
raw_orders = df.to_dict("records")
t.rows_in = len(raw_orders)
t.bytes_processed = df.memory_usage(deep=True).sum()
total_rows_read = t.rows_in
store.log_audit_event(
event_type="RECORD_INSERTED",
target_table="orders",
target_schema="public",
record_count=total_rows_read,
event_metadata={"interval": f"{data_interval_start} to {data_interval_end}"},
)
# --- TRANSFORM ---
valid_orders = []
rejected_orders = []
with store.task("transform_and_validate") as t:
t.rows_in = len(raw_orders)
for order in raw_orders:
# Validate: orders must have a positive total and a known status
if order["total_amount"] <= 0:
rejected_orders.append({**order, "rejection_reason": "non_positive_amount"})
continue
if order["status"] not in ("completed", "shipped", "processing", "cancelled"):
rejected_orders.append({**order, "rejection_reason": f"unknown_status:{order['status']}"})
continue
# Enrich
order["processed_at"] = datetime.now(timezone.utc).isoformat()
order["pipeline_run_id"] = run_id # Track provenance in the data itself
valid_orders.append(order)
t.rows_out = len(valid_orders)
t.rows_rejected = len(rejected_orders)
total_rows_rejected = t.rows_rejected
# Log validation failures as audit events if they're above a threshold
if rejected_orders:
store.log_audit_event(
event_type="VALIDATION_FAILURE",
target_table="fact_orders",
target_schema="analytics",
record_count=len(rejected_orders),
event_metadata={
"sample_rejections": rejected_orders[:5], # First 5 as a sample
"rejection_rate_pct": round(len(rejected_orders) / len(raw_orders) * 100, 2),
},
)
# --- LOAD ---
with store.task("load_to_warehouse") as t:
t.rows_in = len(valid_orders)
if valid_orders:
valid_df = pd.DataFrame(valid_orders)
warehouse_engine = create_engine(WAREHOUSE_URL)
# Upsert logic — simplified here for readability
valid_df.to_sql(
"fact_orders_staging",
warehouse_engine,
schema="analytics",
if_exists="replace",
index=False,
)
with warehouse_engine.connect() as conn:
result = conn.execute(text("""
INSERT INTO analytics.fact_orders
SELECT * FROM analytics.fact_orders_staging
ON CONFLICT (order_id) DO UPDATE
SET total_amount = EXCLUDED.total_amount,
status = EXCLUDED.status,
processed_at = EXCLUDED.processed_at,
pipeline_run_id = EXCLUDED.pipeline_run_id
"""))
conn.commit()
rows_affected = result.rowcount
t.rows_out = rows_affected
total_rows_written = rows_affected
# All tasks complete — mark the run as successful
store.complete_run(
rows_read=total_rows_read,
rows_written=total_rows_written,
rows_rejected=total_rows_rejected,
status="SUCCESS",
)
except Exception as e:
store.fail_run(error_message=str(e))
raise
Notice the pipeline_run_id being written directly into the fact_orders records. This is one of the most powerful practices you can adopt: every row in your warehouse carries the ID of the pipeline run that created or last modified it. When someone asks "where did this order record come from?" you can join directly to your metadata store and answer in seconds.
Tip: Embedding
pipeline_run_idin your target tables also makes reprocessing and backfilling historical data much safer. You can identify every record written by a specific run and overwrite or audit just those records, rather than reasoning about date ranges.
The whole point of building this is to make operational questions answerable quickly. Let's look at the queries you'll actually run.
SELECT
p.pipeline_name,
r.run_start_time,
r.run_status,
r.rows_read,
r.rows_written,
r.rows_rejected,
r.duration_seconds,
r.error_message
FROM pipeline_runs r
JOIN pipelines p USING (pipeline_id)
WHERE r.run_start_time >= '2024-01-13'
AND r.run_start_time < '2024-01-15'
ORDER BY r.run_start_time DESC;
WITH historical_stats AS (
SELECT
pipeline_id,
AVG(rows_written) AS avg_rows,
STDDEV(rows_written) AS stddev_rows,
MIN(rows_written) AS min_rows,
MAX(rows_written) AS max_rows
FROM pipeline_runs
WHERE run_status = 'SUCCESS'
AND run_start_time >= NOW() - INTERVAL '30 days'
AND run_start_time < NOW() - INTERVAL '1 day' -- Exclude today
GROUP BY pipeline_id
),
todays_run AS (
SELECT pipeline_id, rows_written, run_start_time
FROM pipeline_runs
WHERE run_status = 'SUCCESS'
AND run_start_time >= CURRENT_DATE
ORDER BY run_start_time DESC
LIMIT 1
)
SELECT
p.pipeline_name,
t.rows_written AS todays_rows,
ROUND(h.avg_rows) AS historical_avg,
ROUND(h.stddev_rows) AS historical_stddev,
ROUND((t.rows_written - h.avg_rows) / NULLIF(h.stddev_rows, 0), 2) AS z_score,
CASE
WHEN ABS((t.rows_written - h.avg_rows) / NULLIF(h.stddev_rows, 0)) > 3
THEN 'ANOMALOUS'
ELSE 'NORMAL'
END AS assessment
FROM todays_run t
JOIN historical_stats h USING (pipeline_id)
JOIN pipelines p USING (pipeline_id);
A z-score above 3 (or below -3) means today's row count is more than three standard deviations from the 30-day mean. That's a signal worth investigating — either the business had an unusual day, or something in the pipeline changed. This is the kind of lightweight volume-based anomaly detection that catches the "the pipeline ran but processed nothing" scenario from our opening.
SELECT
ae.event_time,
p.pipeline_name,
r.run_id,
ae.event_type,
ae.record_count,
ae.event_metadata->>'rejection_rate_pct' AS rejection_rate,
r.triggered_by
FROM audit_events ae
JOIN pipeline_runs r USING (run_id)
JOIN pipelines p USING (pipeline_id)
WHERE ae.target_table = 'fact_orders'
AND ae.event_time >= NOW() - INTERVAL '7 days'
ORDER BY ae.event_time DESC;
SELECT
p.pipeline_name,
tr.task_name,
ROUND(AVG(tr.duration_seconds)) AS avg_duration_secs,
MAX(tr.duration_seconds) AS max_duration_secs,
ROUND(AVG(tr.rows_out)) AS avg_rows_out,
COUNT(*) AS run_count
FROM task_runs tr
JOIN pipeline_runs r USING (run_id)
JOIN pipelines p USING (pipeline_id)
WHERE tr.task_status = 'SUCCESS'
AND tr.start_time >= NOW() - INTERVAL '14 days'
GROUP BY p.pipeline_name, tr.task_name
ORDER BY avg_duration_secs DESC;
This query is invaluable during performance investigations. If the load_to_warehouse task is taking 4x longer this week than last week, you can see it here before a user complains about SLA breaches. This ties directly into building and managing data pipeline SLAs — your metadata store is the foundation on which SLA measurement is built.
Note: If you're running many pipelines, add a scheduled query (or a dbt model over your metadata store) that materializes these aggregates daily. Querying 90 days of raw
task_runswith millions of rows can be slow — pre-aggregated "pipeline health" summary tables pay off quickly.
Raw queries are powerful, but in production you want automated alerts when things go wrong, not manual dashboard checks. Here's a lightweight alerter that runs as a post-pipeline check or as a standalone scheduled job:
# monitoring/volume_check.py
from datetime import datetime, timezone, timedelta
from sqlalchemy import create_engine, text
def check_volume_thresholds(metadata_db_url: str, pipeline_name: str) -> dict:
"""
Compare today's run volume against 30-day historical average.
Returns a result dict with status and diagnostics.
"""
engine = create_engine(metadata_db_url)
with engine.connect() as conn:
result = conn.execute(text("""
WITH stats AS (
SELECT
AVG(rows_written) AS avg_rows,
STDDEV(rows_written) AS stddev_rows
FROM pipeline_runs r
JOIN pipelines p USING (pipeline_id)
WHERE p.pipeline_name = :name
AND r.run_status = 'SUCCESS'
AND r.run_start_time BETWEEN NOW() - INTERVAL '31 days' AND NOW() - INTERVAL '1 day'
),
latest AS (
SELECT rows_written, run_id, run_start_time
FROM pipeline_runs r
JOIN pipelines p USING (pipeline_id)
WHERE p.pipeline_name = :name
AND r.run_status = 'SUCCESS'
ORDER BY r.run_start_time DESC
LIMIT 1
)
SELECT
l.run_id,
l.rows_written,
s.avg_rows,
s.stddev_rows,
CASE WHEN s.stddev_rows > 0
THEN (l.rows_written - s.avg_rows) / s.stddev_rows
ELSE 0
END AS z_score
FROM latest l, stats s
"""), {"name": pipeline_name}).fetchone()
if not result:
return {"status": "NO_DATA", "message": "No successful runs found."}
z_score = float(result.z_score) if result.z_score else 0
status = "OK"
if abs(z_score) > 3:
status = "CRITICAL"
elif abs(z_score) > 2:
status = "WARNING"
return {
"status": status,
"run_id": str(result.run_id),
"rows_written": result.rows_written,
"historical_avg": round(float(result.avg_rows), 1) if result.avg_rows else None,
"z_score": round(z_score, 2),
"message": (
f"Row count {result.rows_written} is {abs(z_score):.1f} std devs "
f"{'above' if z_score > 0 else 'below'} the 30-day average "
f"({round(float(result.avg_rows), 0):.0f} rows)."
if result.avg_rows else "Insufficient historical data."
),
}
Plug this function into your orchestration layer — after the pipeline completes successfully, run this check, and if it returns CRITICAL or WARNING, send it to your alerting system. This is composable with whatever alerting infrastructure you already have, whether that's PagerDuty, Slack webhooks, or email.
Warning: Don't set static row count thresholds like "alert if fewer than 1,000 rows." Business volumes change over time — what was normal in Q1 isn't normal in Q4. Statistical thresholds based on rolling averages adapt automatically to seasonality and growth, making them far more durable in production.
In this exercise, you'll extend the metadata store to support a pipeline registry pattern that's essential for teams with many pipelines.
The scenario: Your team runs 15 different pipelines, and you want a single "control plane" view that shows the health of every pipeline in one query — last run time, last run status, 7-day success rate, and average row volume.
Your task:
Create the schema from this lesson in a local PostgreSQL or SQLite database (for SQLite, skip the gen_random_uuid() default and generate UUIDs in Python instead).
Write a script that inserts 30 days of synthetic run history for three pipelines: customer_orders_daily, product_catalog_sync, and marketing_attribution_weekly. Include realistic variance in row counts (use Python's random.gauss()) and inject 3-4 failed runs spread across the history.
Write a SQL query that produces a "pipeline health dashboard" view with these columns:
pipeline_namelast_run_timelast_run_statussuccess_rate_7d (percentage of runs in the last 7 days that succeeded)avg_rows_7d (average rows written in successful runs over 7 days)health_indicator — a CASE expression that shows HEALTHY, DEGRADED (success rate between 50-90%), or CRITICAL (below 50%)Extend the MetadataStore client with a get_last_successful_run() method that returns the data_interval_end timestamp of the last successful run. This is the building block for incremental loading patterns where you pick up from where the last run left off.
Stretch goal: Add a pipeline_dependencies table that records which pipelines depend on others (e.g., marketing_attribution_weekly depends on customer_orders_daily). Write a query that identifies "blocked" pipelines — those whose upstream dependency hasn't had a successful run in the expected window. This is the beginning of lineage tracking, which we explore further in understanding data pipeline dependencies.
If every task makes a synchronous database write before proceeding, and your metadata database is slow or overloaded, you've just added latency to every single pipeline. In extreme cases, a metadata store outage can take down all your pipelines.
Fix: Use a write-behind pattern — buffer metadata writes to a local queue (even a simple in-memory list) and flush them asynchronously. Better yet, host your metadata store on a different instance than your production databases so resource contention can't cascade. For very high-volume pipelines, consider batching task metrics into a buffer and flushing at the end of the run.
A common bug: the pipeline fails during the extract task, an exception propagates, and complete_run() never gets called. Now you have a run record stuck in RUNNING status forever.
Fix: Always wrap your entire pipeline in a try/except/finally block and call either complete_run() or fail_run() in the finally clause — not the except clause. The task() context manager in our implementation already does this for tasks; do the same at the pipeline level.
It's tempting to store entire row diffs in the old_value/new_value JSONB columns. For a table with 100 columns, that makes your audit_events table enormous very quickly.
Fix: Be surgical. Store only the fields that are meaningful for audit purposes — the primary key, the changed fields, and the change timestamp. If you need full row diffs, consider a dedicated change data capture system rather than stuffing it into a general-purpose audit table.
If your audit_events table has 10 million rows and your audit query filters on target_table, you're doing a full table scan without an index on that column.
Fix: The schema above includes indexes on target_table and event_time. Make sure you add pipeline_id to the pipeline_runs index (already included), and consider a composite index on (pipeline_id, run_start_time DESC) for the common pattern of "show me the last 30 runs of pipeline X."
If you're using multiprocessing or spawning subprocesses for parallelism, your run_id doesn't automatically cross process boundaries.
Fix: Pass the run_id explicitly as an argument to any child process or worker function. If you're using a framework like Celery or Ray, include the run_id in the task payload. This connects back to the pipeline dependency injection pattern — treat your metadata store client as a dependency that gets explicitly passed, never globally imported.
Tip: Add a
get_run_id()method to yourMetadataStoreclass and document that it should be called before spawning any child processes. Making this an explicit step — rather than relying on global state — prevents the subtle bug where child processes can't write task records because they don't know their parent'srun_id.
You've built a production-grade pipeline metadata store from the ground up. The key capabilities you now have:
pipeline_run_id written into your target tables so every downstream row can be traced back to its origin run.The metadata store you've built here isn't just an operational convenience — it becomes the foundation for several other important patterns. Your SLA monitoring system can query duration_seconds and run_end_time against committed SLA windows. Your checkpointing and state management logic can use get_last_successful_run() to determine exactly where to resume after a failure. Your data quality validation layer can read rejection rates from rows_rejected and escalate when they breach a threshold.
For your next step, explore data quality validation, testing, and monitoring pipelines to see how structured volume metrics from your metadata store can feed into a broader data quality framework — including column-level completeness checks and cross-run consistency validation.
Data Pipeline Fundamentals