Data Quality Monitoring in Production

How to monitor data quality in production pipelines. Covers data contracts, schema validation, anomaly detection, lineage tracking, and building a data quality culture.

Data quality monitoring in production ensures that datasets used by machine learning models, dashboards, and operational systems are accurate, consistent, timely, and fit for purpose. It is essential when data pipelines run autonomously across multiple services, and when downstream systems depend on data that is not manually inspected. It matters when a model’s predictions degrade silently, when a dashboard shows stale metrics, or when a batch job fails to materialize a table due to missing or malformed input.

Detecting Data Drift and Schema Changes

Data drift occurs when the statistical properties of input data change over time. In production, this often goes unnoticed until model performance degrades. The first signal is a change in the distribution of a key feature or the arrival of data with unexpected column names or types.

To detect drift, define a baseline from historical data, then compare incoming batches against it using statistical tests or distance metrics.

A typical workflow uses a tool like Great Expectations or Deequ, but the pattern is consistent across systems:


great_expectations dataset new --name user_engagement --batch-kwargs '{"path": "data/train.csv"}'

# Validate a new batch against the expectations
great_expectations checkpoint run user_engagement_checkpoint --batch-kwargs '{"path": "data/2024-05-03.csv"}'

The output includes a JSON report with validation results, failure counts, and summary statistics. A critical check is expect_column_values_to_not_be_null, which triggers immediately if any required column contains NULL values.

When a batch fails, the system can emit a structured alert:

{
  "checkpoint": "user_engagement_checkpoint",
  "batch_id": "2024-05-03",
  "timestamp": "2024-05-03T12:00:00Z",
  "success": false,
  "details": [
    {
      "expectation": "expect_column_values_to_not_be_null",
      "column": "user_id",
      "result": {
        "success": false,
        "unexpected_count": 142,
        "unexpected_percent": 7.3,
        "unexpected_list": [null, null, null, 12345, null]
      }
    }
  ]
}

This JSON is ingested into a monitoring system like Prometheus or Datadog, where it drives alerting and dashboarding.

Schema changes are detected by comparing the schema of a new batch to a known schema. A common approach uses Avro, Parquet, or JSON Schema.

For a Parquet file, validate schema compatibility using pyarrow:

import pyarrow as pa
import pyarrow.parquet as pq

def validate_schema(new_file, baseline_schema):
    new_table = pq.read_table(new_file)
    new_schema = new_table.schema
    return pa.schema(baseline_schema).equals(new_schema)

# Use in a data pipeline
if not validate_schema("data/events/2024-05-03.parquet", ["user_id", "event_type", "timestamp", "session_duration"]):
    raise RuntimeError("Schema mismatch: new data does not match expected schema")

Schema drift is often accompanied by a metadata change: a new column appears, a column’s type changes (e.g., int64 → double), or a column becomes required. These are detected using a schema registry that stores historical versions.

Ensuring Completeness and Timeliness

Completeness means all expected data is present. Timeliness means data arrives within expected deadlines. These are enforced through time-windowed checks and record-count validations.

A production data pipeline processes daily batches. Each batch must contain at least 10,000 records. The pipeline writes to a table daily_metrics and logs the count in a batch_metadata table.

CREATE TABLE batch_metadata (
    batch_id TEXT PRIMARY KEY,
    batch_date DATE,
    record_count INT,
    status TEXT,
    processed_at TIMESTAMPTZ,
    validation_errors JSONB
);

A cron job runs every hour to validate that the expected batch has been processed:

#!/bin/bash
# check_completeness.sh

BATCH_DATE=$(date -d "yesterday" +%Y-%m-%d)
EXPECTED_COUNT=10000

COUNT=$(psql -c "
  SELECT count(*) FROM daily_metrics
  WHERE date = '$BATCH_DATE'
  AND processed_at >= now() - interval '1 hour'
" -t)

if [ "$COUNT" -lt "$EXPECTED_COUNT" ]; then
  echo "Incomplete batch: $COUNT records, expected $EXPECTED_COUNT" >&2
  # Write to metadata table
  psql << EOF
INSERT INTO batch_metadata (batch_id, batch_date, record_count, status, processed_at)
VALUES ('daily-$BATCH_DATE', '$BATCH_DATE', $COUNT, 'incomplete', now())
ON CONFLICT (batch_id) DO UPDATE SET record_count = EXCLUDED.record_count, status = 'incomplete', processed_at = now();
EOF
fi

This script runs via systemd:

# /etc/systemd/system/data-completeness-check.service
[Unit]
Description=Check completeness of daily data batches
After=network.target

[Service]
Type=oneshot
ExecStart=/usr/local/bin/check_completeness.sh
User=datapipeline
Environment=PATH=/usr/local/bin:/usr/bin:/bin

[Install]
WantedBy=multi-user.target

Timeliness is validated by checking the processed_at timestamp. A batch is considered late if it arrives more than 6 hours after the expected window.

A failure appears in the logs:

2024-05-03T12:30:00Z ERROR: Incomplete batch: 8214 records, expected 10000
2024-05-03T12:30:00Z ERROR: Batch daily-2024-05-03 is late: processed_at = 2024-05-03 05:12:44, expected by 2024-05-03 06:00:00

These messages are collected by Fluentd and routed to Elasticsearch for alerting.

Monitoring for Data Freshness and Latency

Freshness measures when the latest data was ingested. Latency measures how long it takes from data generation to availability.

A real-time ingestion pipeline uses Kafka to stream events. Each event includes a timestamp field. The latest event should arrive within 15 minutes of its timestamp.

A freshness check compares the max(timestamp) of the Kafka topic to the current time:

#!/bin/bash
# check_freshness.sh

TOPIC="user_events"
MAX_TIMESTAMP=$(kafka-topics.sh --bootstrap-server kafka:9092 \
  --describe --topic $TOPIC | grep -A 1 "Leader" | tail -1 | cut -d' ' -f1)

if [ -z "$MAX_TIMESTAMP" ]; then
  echo "No events in topic $TOPIC" >&2
  exit 1
fi

# Convert Kafka timestamp (epoch in ms) to ISO 8601
LATEST_TIME=$(date -d "@$((MAX_TIMESTAMP / 1000))" +"%Y-%m-%dT%H:%M:%SZ")

NOW=$(date -u +"%Y-%m-%dT%H:%M:%SZ")

# Compare with 15-minute threshold
if [ $(($(date -d "$NOW" +%s) - $(date -d "$LATEST_TIME" +%s))) -gt 900 ]; then
  echo "Freshness alert: latest event at $LATEST_TIME, now is $NOW"
  echo "Latency exceeds 15 minutes" >&2
  # Send alert via HTTP
  curl -X POST http://alerting-service:8080/api/alert \
    -H "Content-Type: application/json" \
    -d '{
      "title": "Kafka freshness alert",
      "message": "Latest event at ' + $LATEST_TIME + ', now is ' + $NOW + '.",
      "severity": "warning",
      "service": "user_events"
    }'
fi

This script runs every 5 minutes via cron.

Latency is measured end-to-end. A pipeline emits events at 08:00:00 UTC. The ingestion system logs the processed_at time when it writes the data to a database. The latency is calculated as processed_at - timestamp.

A latency alert triggers when the 95th percentile latency exceeds 30 minutes:

SELECT
  percentile_disc(0.95) WITHIN GROUP (ORDER BY processed_at - timestamp) as latency_95
FROM user_events
WHERE date = '2024-05-03';

When the query returns a value above 30 minutes, a monitoring system sends an alert.

Handling Outliers and Anomalies

Outliers are data points that deviate significantly from the expected range. Anomalies are unexpected patterns over time, such as sudden drops in volume or spikes in a metric.

A common approach is to define thresholds using historical statistics.

For example, a metric total_revenue is expected to be between 10,000 and 200,000 per day. A threshold check uses the interquartile range (IQR):

import numpy as np

def detect_outliers(series, multiplier=1.5):
    Q1 = np.percentile(series, 25)
    Q3 = np.percentile(series, 75)
    IQR = Q3 - Q1
    lower_bound = Q1 - multiplier * IQR
    upper_bound = Q3 + multiplier * IQR
    outliers = [x for x in series if x < lower_bound or x > upper_bound]
    return outliers, lower_bound, upper_bound

# Use in a pipeline
revenue = [12000, 15000, 18000, 21000, 25000, 30000, 50000, 100000, 120000]
outliers, lb, ub = detect_outliers(revenue)
if outliers:
    print(f"Outliers detected: {outliers}")
    print(f"Expected range: {lb:.2f} to {ub:.2f}")

Anomalies are detected using time series analysis. A simple method uses Z-scores:

def detect_anomalies_zscore(timeseries, window=7, threshold=2.0):
    rolling_mean = timeseries.rolling(window=window).mean()
    rolling_std = timeseries.rolling(window=window).std()
    z_scores = (timeseries - rolling_mean) / rolling_std
    anomalies = z_scores[abs(z_scores) > threshold]
    return anomalies

# Example
daily_revenue = pd.read_csv("daily_revenue.csv", index_col="date", parse_dates=True)
anomalies = detect_anomalies_zscore(daily_revenue["revenue"], window=7)

The output is a list of dates where the revenue deviated significantly from the moving average.

Anomalies are stored in a table:

CREATE TABLE anomalies (
    id SERIAL PRIMARY KEY,
    metric TEXT,
    date DATE,
    value DOUBLE PRECISION,
    expected_value DOUBLE PRECISION,
    z_score DOUBLE PRECISION,
    detected_at TIMESTAMPTZ DEFAULT now(),
    metadata JSONB
);

Each anomaly is linked to a dashboard in Grafana or Superset, where users can drill down into the data and investigate root causes.

Configuring and Operating a Data Quality Pipeline

Data quality monitoring is not a one-off task. It must be configured and operated as part of the data platform.

The core configuration is stored in YAML, with a standard schema:

# data_quality_config.yaml
checkpoints:
  - name: user_engagement
    source: "s3://data/2024-05-03.csv"
    expectations:
      - type: column_values_to_not_be_null
        column: user_id
        threshold: 0.95
      - type: column_mean_to_be_between
        column: session_duration
        lower: 120
        upper: 300
      - type: column_value_set
        column: event_type
        values: [click, view, purchase, checkout]
    alerting:
      channels:
        - type: email
          recipients: ["[email protected]"]
        - type: slack
          webhook_url: "https://hooks.slack.com/services/..."
      on_failure: "send_alert"
      on_success: "log_to_monitoring"

This file is versioned in Git and deployed via CI/CD.

A failure mode common in production is misconfigured thresholds. A checkpoint expects user_id to be non-null in at least 95% of rows, but the pipeline uses a 90% threshold. When the data arrives with 92% non-null values, the checkpoint passes, but the underlying data is still problematic.

Another silent failure: schema evolution without backward compatibility. A new version of a schema adds a column user_tier but does not populate it. The data quality system detects the column but does not validate its values. The column is present but full of NULLs, which go unnoticed.

The system must also handle gradual rollouts. A new data format is introduced, but both old and new formats coexist. The monitoring system must detect mismatches between formats and flag them as warnings.

Common Pitfalls and Sharp Edges

Data quality monitoring in production is not a set of checks. It is a living system that evolves with data, driven by metrics, alerting, and configuration. It is the first line of defense against silent data failures.

This page was rewritten on 17 October 2026. It replaced a templated version that was not specific to its own title. How these pages are written and checked. If anything here is wrong, tell us at [email protected] and we will correct it.