Catch bad data before the business does - data profiling monitors, drift metrics and alerts on Unity Catalog tables

The goal: know that gold.sales.orders arrived with 40% null customer_ids, or half the usual row count, before a dashboard user tells you. Pipeline expectations catch rules you already know. Data profiling catches the changes you didn't think to write a rule for: null rates creeping up, a category disappearing, a numeric column shifting.

This post sets up a profile on a table with the Python SDK, reads the metric tables, and wires alerts on-call will actually see.

#How it works

• A monitor (data profile) is attached to one Unity Catalog Delta table. Only one profile per table per metastore.
• Each refresh runs on serverless compute and writes two tables to an output schema:
  • <table>_profile_metrics: per window, per column: count, num_nulls, percent_null, distinct_count, min, max, avg, quantiles, frequent_items and more.
  • <table>_drift_metrics: change between windows (CONSECUTIVE) or against a baseline table (BASELINE): count_delta, percent_null_delta, ks_test, js_distance, population_stability_index and more.
• A dashboard is generated in the assets_dir you choose.

Note

The feature was called Lakehouse Monitoring and the SDK class was w.quality_monitors. That class is deprecated in favour of w.data_quality (REST /api/data-quality/v1/monitors). Names in the UI and docs vary by release; check your SDK version if an import fails.

#Prerequisites and checks

1. Unity Catalog table (managed or external Delta) and serverless compute enabled in the workspace.
2. An output schema for the metric tables, for example ops.dq.
3. The person or service principal creating the monitor needs to manage the table (owner, or MANAGE plus USE CATALOG/USE SCHEMA) and CREATE TABLE on the output schema.
4. For TimeSeries, turn on change data feed so refreshes process only new data:

CREATE SCHEMA IF NOT EXISTS ops.dq;
ALTER TABLE gold.sales.orders SET TBLPROPERTIES (delta.enableChangeDataFeed = true);

#Create a time series profile

Run in a notebook or as a job task:

%pip install "databricks-sdk>=0.68.0"
dbutils.library.restartPython()

from databricks.sdk import WorkspaceClient
from databricks.sdk.service.dataquality import (
    Monitor, DataProfilingConfig, TimeSeriesConfig, AggregationGranularity,
)

w = WorkspaceClient()
table = w.tables.get(full_name="gold.sales.orders")
out_schema = w.schemas.get(full_name="ops.dq")

config = DataProfilingConfig(
    output_schema_id=out_schema.schema_id,
    assets_dir="/Workspace/Shared/dq/gold.sales.orders",
    time_series=TimeSeriesConfig(
        timestamp_column="order_ts",
        granularities=[AggregationGranularity.AGGREGATION_GRANULARITY_1_DAY],
    ),
    slicing_exprs=["region", "channel"],
)

monitor = w.data_quality.create_monitor(
    monitor=Monitor(object_type="table", object_id=table.table_id,
                    data_profiling_config=config)
)
print(monitor)

For a dimension table, replace time_series=... with snapshot=SnapshotConfig() (import SnapshotConfig from the same module).

Tip

Keep slicing_exprs to a few low-cardinality columns. Every slice value gets its own set of metrics, so slicing on customer_id multiplies refresh cost and metric table size.

#Refresh after each load, not on a fixed clock

A schedule works, but a refresh right after the load means metrics describe the data you just wrote. Add this as the last task of the load job:

from databricks.sdk import WorkspaceClient
from databricks.sdk.service.dataquality import Refresh

w = WorkspaceClient()
tid = w.tables.get(full_name="gold.sales.orders").table_id

run = w.data_quality.create_refresh(
    object_type="table", object_id=tid,
    refresh=Refresh(object_type="table", object_id=tid),
)
print(run)

Check recent refreshes and their state:

for r in w.data_quality.list_refresh(object_type="table", object_id=tid):
    print(r.refresh_id, r.state, r.start_time_ms, r.end_time_ms)

Field names on the refresh object can differ by SDK version; print one object to see what yours returns.

#Query the metric tables

Table-level row counts per day (column_name = ':table', whole table = slice_key IS NULL):

SELECT window.start AS day, count AS row_count
FROM ops.dq.orders_profile_metrics
WHERE column_name = ':table'
  AND slice_key IS NULL
  AND log_type = 'INPUT'
ORDER BY day DESC
LIMIT 14;

Columns whose null rate jumped against the previous window:

SELECT window.start AS day, column_name, percent_null_delta
FROM ops.dq.orders_drift_metrics
WHERE drift_type = 'CONSECUTIVE'
  AND slice_key IS NULL
  AND column_name <> ':table'
  AND percent_null_delta > 5
ORDER BY day DESC, percent_null_delta DESC;

Numeric distribution drift (KS test p-value) and categorical drift (JS distance):

SELECT window.start AS day, column_name,
       ks_test.pvalue AS ks_pvalue,
       js_distance
FROM ops.dq.orders_drift_metrics
WHERE drift_type = 'CONSECUTIVE'
  AND slice_key IS NULL
  AND (ks_test.pvalue < 0.01 OR js_distance > 0.2)
ORDER BY day DESC;

The thresholds are starting points. Tune them per table after a couple of weeks of history, or you'll get daily false alarms.

#Turn metrics into on-call alerts

Create a Databricks SQL query that returns one row per problem, then a SQL alert on it.

-- dq_orders_today: returns rows only when something is wrong
WITH today AS (
  SELECT count AS rows_today FROM ops.dq.orders_profile_metrics
  WHERE column_name = ':table' AND slice_key IS NULL AND log_type = 'INPUT'
  ORDER BY window.start DESC LIMIT 1
),
baseline AS (
  SELECT avg(count) AS rows_avg FROM (
    SELECT count FROM ops.dq.orders_profile_metrics
    WHERE column_name = ':table' AND slice_key IS NULL AND log_type = 'INPUT'
    ORDER BY window.start DESC LIMIT 8 OFFSET 1)
)
SELECT 'row_count_drop' AS issue, rows_today, round(rows_avg) AS rows_avg
FROM today, baseline
WHERE rows_today < 0.5 * rows_avg;

Set the alert condition to "row count > 0", schedule it after the refresh task, and send it to a notification destination wired to your paging tool (Teams, PagerDuty or a webhook) rather than a personal mailbox. Add a second alert on refresh failures using NotificationSettings(on_failure=...) in the profile config, or by failing the job task when create_refresh ends in a failed state.

#Symptom → cause → fix

#Removing a monitor

w.data_quality.delete_monitor(object_type="table", object_id=tid)

This doesn't drop the metric tables or the dashboard. Drop them yourself once you no longer need the history:

DROP TABLE IF EXISTS ops.dq.orders_profile_metrics;
DROP TABLE IF EXISTS ops.dq.orders_drift_metrics;

#Prevent it next time

• Profile the 10-20 gold tables that feed decisions, not every bronze table.
• Create monitors from code (a job or Asset Bundle task) so they're recreated in each environment, and run them as a service principal.
• Trigger refreshes from the load job and alert on both data problems and refresh failures.
• Review thresholds monthly and keep pipeline expectations for known rules. Profiling complements expectations; it doesn't replace them.