Databricks Pipeline Monitoring Framework
How We Built a Custom Monitoring Framework for Databricks Pipelines
Data pipelines fail silently. That’s the uncomfortable truth most data engineering teams learn the hard way — usually when a business stakeholder asks why last week’s report looks off, or why a dashboard hasn’t refreshed since Tuesday. By then, the damage is done.
Rather than reaching for an expensive third-party observability tool, we built something lean, purposeful, and entirely native to our Databricks environment. This post walks through what I built, why I built it that way, and the key design decisions that made it work.
The Problem With “Just Check the Logs”
When a pipeline fails, Databricks marks the job as failed. But operational failure is only half the story. A pipeline can succeed — finishing on time, writing rows, returning exit code 0 — while quietly producing garbage data. Null primary keys. Negative claim amounts. Row counts that dropped 40% with no explanation.
Traditional job monitoring doesn’t catch that. You need a second layer: data quality monitoring. And you need both layers to speak the same language so you can ask: “For every run that breached our SLO in the last 30 days, which quality checks also failed?”
That was our north star.
Two Tables, One Foreign Key
The entire framework rests on two Delta tables in Unity Catalog:
pipeline_metrics captures operational health — one row per pipeline run.
pipeline_quality_checks captures data quality — one row per check per run.
Both share a run_id column — a UUID generated at the start of every run. That single foreign key is what makes the whole system queryable.
Section 1 — Configuration
Set these values once per environment. All secrets live in Databricks Secret Scopes — never hardcoded.
# Catalog and schema for monitoring tables
MONITORING_CATALOG = "monitoring"
MONITORING_SCHEMA = "pipeline_ops"
# Full table names
METRICS_TABLE = f"{MONITORING_CATALOG}.{MONITORING_SCHEMA}.pipeline_metrics"
QUALITY_TABLE = f"{MONITORING_CATALOG}.{MONITORING_SCHEMA}.pipeline_quality_checks"
# Slack webhook — stored in Databricks Secrets
SLACK_WEBHOOK_URL = dbutils.secrets.get(scope="monitoring", key="slack_webhook_url")
Section 2 — Table Setup
Call setup_monitoring_tables() once when deploying to a new environment. It is idempotent — safe to re-run because IF NOT EXISTS prevents overwrites.
def setup_monitoring_tables():
# pipeline_metrics — one row per run
spark.sql(f"""
CREATE TABLE IF NOT EXISTS {METRICS_TABLE} (
run_id STRING,
pipeline_name STRING,
pipeline_version STRING,
started_at TIMESTAMP,
completed_at TIMESTAMP,
duration_minutes DOUBLE,
rows_processed LONG,
rows_quarantined LONG,
slo_target_minutes INT,
slo_met BOOLEAN,
status STRING,
error_message STRING
) USING DELTA
""")
# pipeline_quality_checks — one row per check per run
spark.sql(f"""
CREATE TABLE IF NOT EXISTS {QUALITY_TABLE} (
run_id STRING,
pipeline_name STRING,
checked_at TIMESTAMP,
check_name STRING,
column_name STRING,
check_type STRING,
rows_checked LONG,
rows_failed LONG,
failure_pct DOUBLE,
threshold_pct DOUBLE,
passed BOOLEAN
) USING DELTA
""")
Section 3 — Pipeline Metrics Writer
Call write_pipeline_metrics() at the end of every run — in both the success path and the except block. The completed_at timestamp is captured inside the function so it always reflects when the work actually finished.
def write_pipeline_metrics(
pipeline_name, started_at, rows_processed,
rows_quarantined, run_id, pipeline_version='1.0',
slo_target_minutes=45, status='success', error_message=None
):
completed_at = datetime.utcnow()
duration_minutes = (completed_at - started_at).total_seconds() / 60
slo_met = duration_minutes <= slo_target_minutes
row = Row(
run_id = run_id,
pipeline_name = pipeline_name,
pipeline_version = pipeline_version,
started_at = started_at,
completed_at = completed_at,
duration_minutes = round(duration_minutes, 2),
rows_processed = rows_processed,
rows_quarantined = rows_quarantined,
slo_target_minutes = slo_target_minutes,
slo_met = slo_met,
status = status,
error_message = error_message
)
# Append only — this table is an audit log, never overwrite
spark.createDataFrame([row]) \
.write.format('delta').mode('append').saveAsTable(METRICS_TABLE)
return duration_minutes, slo_met
Section 4 — Quality Check Runner
Call run_quality_check() once per check. Each call appends one row to pipeline_quality_checks. The condition argument is a PySpark Column expression that evaluates to True for rows that fail the check.
def run_quality_check(
df, run_id, pipeline_name, check_name,
column_name, check_type, condition, threshold_pct=1.0
):
rows_checked = df.count()
rows_failed = df.filter(condition).count()
failure_pct = (rows_failed / rows_checked * 100) if rows_checked > 0 else 0.0
passed = failure_pct <= threshold_pct
spark.createDataFrame([Row(
run_id = run_id,
pipeline_name = pipeline_name,
checked_at = datetime.utcnow(),
check_name = check_name,
column_name = column_name,
check_type = check_type,
rows_checked = rows_checked,
rows_failed = rows_failed,
failure_pct = round(failure_pct, 4),
threshold_pct = threshold_pct,
passed = passed
)]).write.format('delta').mode('append').saveAsTable(QUALITY_TABLE)
return passed
Example Checks
# Null check — zero tolerance on primary key
run_quality_check(
df=silver_df, run_id=run_id,
pipeline_name=PIPELINE_NAME, check_name='patient_id_not_null',
column_name='patient_id', check_type='null_check',
condition=col('patient_id').isNull(), threshold_pct=0.0
)
# Range check — claim amounts must be positive
run_quality_check(
df=silver_df, run_id=run_id,
pipeline_name=PIPELINE_NAME, check_name='claim_amount_positive',
column_name='claim_amount', check_type='range_check',
condition=col('claim_amount') <= 0, threshold_pct=0.5
)
# Volume check — row count must not drop vs last run
run_quality_check(
df=silver_df, run_id=run_id,
pipeline_name=PIPELINE_NAME, check_name='volume_vs_last_run',
column_name='*', check_type='volume_check',
condition=lit(rows_processed < expected_min), threshold_pct=0.0
)
Section 5 — The Pipeline Wrapper
This is the pattern used in every pipeline notebook. It ties together start time, run_id, transformation logic, quality checks, metrics writing, and alerting in a single try/except block.
start_time = datetime.utcnow() # capture before any work
run_id = str(uuid.uuid4()) # shared key between both tables
rows_processed = 0
rows_quarantined = 0
quality_failures = []
try:
# STEP 1 — READ
bronze_df = spark.table('bronze.claims')
rows_processed = bronze_df.count()
# STEP 2 — TRANSFORM
silver_df = transform_bronze_to_silver(bronze_df)
# STEP 3 — QUALITY CHECKS (before writing)
if not run_quality_check(...): # null check
quality_failures.append('patient_id_not_null')
if not run_quality_check(...): # range check
quality_failures.append('claim_amount_positive')
# STEP 4 — WRITE (split passing and failing rows)
silver_df.filter(col('quality_flag') == 'pass') \
.write.format('delta').mode('append').saveAsTable('silver.claims')
silver_df.filter(col('quality_flag') == 'fail') \
.write.format('delta').mode('append').saveAsTable('silver.claims_quarantine')
# STEP 5 — WRITE METRICS
duration, slo_met = write_pipeline_metrics(
pipeline_name = PIPELINE_NAME,
started_at = start_time,
rows_processed = rows_processed,
rows_quarantined = rows_quarantined,
run_id = run_id,
slo_target_minutes = 45,
status = 'success'
)
# STEP 6 — ALERT if needed
if not slo_met:
send_alert(PIPELINE_NAME, run_id, f'SLO breached: {duration:.1f} min')
if quality_failures:
send_alert(PIPELINE_NAME, run_id, f'Checks failed: {quality_failures}')
except Exception as e:
# Still write metrics so the failure is recorded
write_pipeline_metrics(..., status='failed', error_message=str(e))
send_alert(PIPELINE_NAME, run_id, f'Pipeline FAILED: {str(e)}')
raise # re-raise so Databricks marks the job as failed
Section 6 — Monitoring Queries
SLO Summary — Last 7 Days
SELECT
pipeline_name,
COUNT(*) AS total_runs,
SUM(CASE WHEN slo_met THEN 1 ELSE 0 END) AS slo_met_count,
SUM(CASE WHEN NOT slo_met THEN 1 ELSE 0 END) AS slo_breach_count,
ROUND(AVG(duration_minutes), 1) AS avg_duration_min
FROM monitoring.pipeline_ops.pipeline_metrics
WHERE started_at >= CURRENT_DATE - INTERVAL 7 DAYS
GROUP BY pipeline_name
ORDER BY slo_breach_count DESC;
Full Picture for One Run — Operational + Quality
SELECT
m.pipeline_name,
m.duration_minutes,
m.slo_met,
q.check_name,
q.failure_pct,
q.passed
FROM monitoring.pipeline_ops.pipeline_metrics m
JOIN monitoring.pipeline_ops.pipeline_quality_checks q
ON m.run_id = q.run_id
WHERE m.pipeline_name = 'silver_transform_claims'
AND DATE(m.started_at) = CURRENT_DATE
ORDER BY q.passed ASC; -- failed checks first
Quarantine Rate Trend — Last 30 Days
SELECT
pipeline_name,
DATE(started_at) AS run_date,
SUM(rows_quarantined) AS quarantined,
ROUND(SUM(rows_quarantined) * 100.0
/ NULLIF(SUM(rows_processed), 0), 2) AS quarantine_pct
FROM monitoring.pipeline_ops.pipeline_metrics
WHERE started_at >= CURRENT_DATE - INTERVAL 30 DAYS
GROUP BY pipeline_name, DATE(started_at)
ORDER BY run_date DESC;
Key Rules to Live By
- One shared metrics table for all pipelines — pipeline_name separates them
- Use the same run_id in both tables — it is the foreign key linking operational and quality data
- Always write metrics in the except block too — a failed run still needs a record
- Always re-raise after writing failure metrics — so Databricks marks the job as failed
- Store all secrets in Databricks Secret Scopes — never hardcode credentials
- Bump pipeline_version on significant logic changes — enables before/after comparisons
- Set threshold_pct = 0.0 for zero-tolerance checks like primary keys
- Alert on SLO breach and quality failures separately — they have different on-call responses
Closing Thoughts
Data pipelines are only as trustworthy as your ability to verify them. This framework doesn’t try to replace full observability platforms — it gives Databricks teams a structured, queryable, audit-grade foundation that takes an afternoon to deploy and pays dividends every time something goes wrong.
If you’re running pipelines in Databricks today without structured monitoring, this is a very good place to start.
Write metrics on failure, not just success. An unrecorded failure is a blind spot. Always write to the metrics table in the except block.
Re-raise after logging. Your monitoring framework should record what happened. Your orchestrator should know the job failed. These are separate responsibilities.
Make thresholds explicit. A quality check with no documented tolerance is a quality check waiting to cause confusion. Store the threshold alongside the result.
Name pipelines consistently. A naming convention like silver_transform_claims makes filtering by layer trivial and keeps dashboards readable as the number of pipelines grows.
Data pipelines are only as trustworthy as your ability to verify them. A monitoring framework doesn’t have to be complex to be effective — it has to be consistent, complete, and queryable. This one is all three.
If you’re running pipelines in Databricks today without structured monitoring, this is a good place to start.