Data Quality Monitoring

Data quality monitoring ensures the data feeding ML systems remains reliable and consistent. Bad data causes bad predictions - garbage in, garbage out. Unlike model monitoring which catches model degradation, data quality monitoring catches problems at the source before they propagate through pipelines.

Where It's Used

Data quality is critical wherever ML systems consume data:

• Feature pipelines: Validating inputs to feature stores
• Training pipelines: Ensuring training data quality
• Inference pipelines: Validating real-time inputs
• Data warehouses: Monitoring source data quality
• ETL processes: Catching issues during transformation
• Data contracts: Enforcing agreements between teams

Products: Great Expectations, dbt tests, Monte Carlo, Soda, Anomalo.

Data Quality Dimensions

Typical Monitoring Points

What You Actually Do

1. Map data flows - Understand sources, transformations, consumers
2. Define expectations - What should each dataset look like
3. Implement schema validation - Catch structural changes
4. Add statistical checks - Distributions, ranges, correlations
5. Set up freshness monitoring - Data arriving on time
6. Configure alerting - Right people, right severity
7. Build data contracts - Agreements between producers/consumers
8. Create dashboards - Visibility into data health
9. Automate remediation - Where safe to do so
10. Review and iterate - Update expectations as data evolves

Schema Validation

from dataclasses import dataclass
from typing import List, Dict, Any, Optional
import pandas as pd

@dataclass
class ColumnSchema:
    name: str
    dtype: str
    nullable: bool = True
    min_value: Optional[float] = None
    max_value: Optional[float] = None
    allowed_values: Optional[List[Any]] = None

class SchemaValidator:
    def __init__(self, columns: List[ColumnSchema]):
        self.columns = {col.name: col for col in columns}

    def validate(self, df: pd.DataFrame) -> Dict[str, List[str]]:
        errors = {"missing_columns": [], "extra_columns": [], 
                  "type_errors": [], "constraint_errors": []}

        # Check for missing/extra columns
        expected = set(self.columns.keys())
        actual = set(df.columns)
        errors["missing_columns"] = list(expected - actual)
        errors["extra_columns"] = list(actual - expected)

        # Validate each column
        for col_name, schema in self.columns.items():
            if col_name not in df.columns:
                continue

            col = df[col_name]

            # Type check
            if not self._check_dtype(col, schema.dtype):
                errors["type_errors"].append(
                    f"{col_name}: expected {schema.dtype}, got {col.dtype}"
                )

            # Null check
            if not schema.nullable and col.isna().any():
                errors["constraint_errors"].append(
                    f"{col_name}: contains nulls but nullable=False"
                )

            # Range check
            if schema.min_value is not None:
                if (col < schema.min_value).any():
                    errors["constraint_errors"].append(
                        f"{col_name}: values below minimum {schema.min_value}"
                    )
            if schema.max_value is not None:
                if (col > schema.max_value).any():
                    errors["constraint_errors"].append(
                        f"{col_name}: values above maximum {schema.max_value}"
                    )

            # Allowed values check
            if schema.allowed_values is not None:
                invalid = ~col.isin(schema.allowed_values + [None])
                if invalid.any():
                    errors["constraint_errors"].append(
                        f"{col_name}: contains values not in allowed set"
                    )

        return errors

    def _check_dtype(self, series: pd.Series, expected: str) -> bool:
        dtype_map = {
            "int": ["int64", "int32", "Int64"],
            "float": ["float64", "float32"],
            "string": ["object", "string"],
            "bool": ["bool", "boolean"],
            "datetime": ["datetime64[ns]"]
        }
        return str(series.dtype) in dtype_map.get(expected, [expected])

Statistical Monitoring

import numpy as np
from scipy import stats
from typing import Dict, Tuple

class StatisticalMonitor:
    def __init__(self, reference_stats: Dict[str, Dict]):
        self.reference = reference_stats

    @staticmethod
    def compute_stats(df, columns: list) -> Dict[str, Dict]:
        """Compute reference statistics from baseline data."""
        stats_dict = {}
        for col in columns:
            if df[col].dtype in ['float64', 'int64']:
                stats_dict[col] = {
                    "mean": df[col].mean(),
                    "std": df[col].std(),
                    "min": df[col].min(),
                    "max": df[col].max(),
                    "p5": df[col].quantile(0.05),
                    "p95": df[col].quantile(0.95),
                    "null_rate": df[col].isna().mean()
                }
            else:
                value_counts = df[col].value_counts(normalize=True)
                stats_dict[col] = {
                    "cardinality": df[col].nunique(),
                    "null_rate": df[col].isna().mean(),
                    "value_distribution": value_counts.to_dict()
                }
        return stats_dict

    def check_drift(self, df, threshold: float = 0.1) -> Dict[str, Dict]:
        """Compare current data to reference statistics."""
        alerts = {}

        for col, ref_stats in self.reference.items():
            if col not in df.columns:
                alerts[col] = {"error": "column missing"}
                continue

            if "mean" in ref_stats:  # Numeric column
                current_mean = df[col].mean()
                current_std = df[col].std()
                ref_mean = ref_stats["mean"]
                ref_std = ref_stats["std"]

                # Z-score of mean shift
                if ref_std > 0:
                    mean_shift = abs(current_mean - ref_mean) / ref_std
                    if mean_shift > threshold * 3:
                        alerts[col] = {
                            "type": "mean_shift",
                            "reference": ref_mean,
                            "current": current_mean,
                            "z_score": mean_shift
                        }

                # Null rate change
                current_null = df[col].isna().mean()
                if abs(current_null - ref_stats["null_rate"]) > threshold:
                    alerts[col] = alerts.get(col, {})
                    alerts[col]["null_rate_change"] = {
                        "reference": ref_stats["null_rate"],
                        "current": current_null
                    }

            else:  # Categorical column
                current_card = df[col].nunique()
                if abs(current_card - ref_stats["cardinality"]) / ref_stats["cardinality"] > threshold:
                    alerts[col] = {
                        "type": "cardinality_change",
                        "reference": ref_stats["cardinality"],
                        "current": current_card
                    }

        return alerts

    def psi(self, expected: np.ndarray, actual: np.ndarray, bins: int = 10) -> float:
        """Population Stability Index."""
        breakpoints = np.percentile(expected, np.linspace(0, 100, bins + 1))
        breakpoints[0] = -np.inf
        breakpoints[-1] = np.inf

        expected_counts = np.histogram(expected, breakpoints)[0] / len(expected)
        actual_counts = np.histogram(actual, breakpoints)[0] / len(actual)

        # Avoid division by zero
        expected_counts = np.clip(expected_counts, 0.001, None)
        actual_counts = np.clip(actual_counts, 0.001, None)

        psi_value = np.sum((actual_counts - expected_counts) * 
                          np.log(actual_counts / expected_counts))
        return psi_value

Freshness Monitoring

from datetime import datetime, timedelta
from typing import Optional

class FreshnessMonitor:
    def __init__(self, expected_frequency: timedelta, 
                 max_delay: timedelta):
        self.expected_frequency = expected_frequency
        self.max_delay = max_delay
        self.last_arrival = None

    def record_arrival(self, timestamp: datetime = None):
        """Record data arrival."""
        self.last_arrival = timestamp or datetime.utcnow()

    def check_freshness(self) -> dict:
        """Check if data is fresh."""
        now = datetime.utcnow()

        if self.last_arrival is None:
            return {
                "status": "error",
                "message": "No data arrivals recorded"
            }

        age = now - self.last_arrival
        expected_next = self.last_arrival + self.expected_frequency

        if now > expected_next + self.max_delay:
            return {
                "status": "stale",
                "age": str(age),
                "expected_by": expected_next.isoformat(),
                "delay": str(now - expected_next)
            }
        elif now > expected_next:
            return {
                "status": "delayed",
                "age": str(age),
                "delay": str(now - expected_next)
            }
        else:
            return {
                "status": "fresh",
                "age": str(age)
            }

Data Contracts

from dataclasses import dataclass
from typing import List, Dict

@dataclass
class DataContract:
    """Agreement between data producer and consumer."""
    name: str
    producer: str
    consumers: List[str]
    schema: List[ColumnSchema]
    freshness_sla: timedelta
    quality_thresholds: Dict[str, float]

class ContractEnforcer:
    def __init__(self, contracts: List[DataContract]):
        self.contracts = {c.name: c for c in contracts}

    def validate_contract(self, contract_name: str, df, 
                          arrival_time: datetime) -> Dict:
        """Validate data against its contract."""
        contract = self.contracts[contract_name]
        results = {"contract": contract_name, "violations": []}

        # Schema validation
        validator = SchemaValidator(contract.schema)
        schema_errors = validator.validate(df)
        for error_type, errors in schema_errors.items():
            if errors:
                results["violations"].append({
                    "type": "schema",
                    "details": {error_type: errors}
                })

        # Quality checks
        null_rate = df.isna().mean().mean()
        if null_rate > contract.quality_thresholds.get("max_null_rate", 0.1):
            results["violations"].append({
                "type": "quality",
                "metric": "null_rate",
                "value": null_rate,
                "threshold": contract.quality_thresholds["max_null_rate"]
            })

        results["valid"] = len(results["violations"]) == 0
        return results

Evaluation Metrics

Common Mistakes & Checklist

• [ ] Monitoring too late: Catching issues after model uses bad data
• [ ] Alert fatigue: Too many alerts, real issues ignored
• [ ] Static thresholds: Not adapting to seasonal patterns
• [ ] Missing freshness: Not monitoring data timeliness
• [ ] Schema drift ignored: New columns break pipelines silently

Production Checklist

Before deployment: - [ ] Baseline statistics computed - [ ] Schema expectations defined - [ ] Freshness SLAs established - [ ] Alerting configured

After deployment: - [ ] Dashboard visibility - [ ] Alert response procedures - [ ] Regular threshold review - [ ] Contract compliance tracking

Tools & Stack

Mini Case Example

Scenario: Data quality for e-commerce recommendation system.

Approach: - Schema validation on ingestion - Statistical monitoring for price, inventory - 15-minute freshness SLA - Data contracts with upstream teams

Results: - 98% issue detection before model impact - 4 minute average detection time - 75% reduction in data incidents

Related Pages

• Index: ML Field Guide
• Feature Stores
• Model Monitoring
• Data Labeling

Further Reading

• Great Expectations Docs
• Data Contracts Specification
• Fundamentals of Data Engineering

Try it live

Everything above runs in your browser — open Building the Data Foundation for Machine Learning: A Data Quality Score Framework and change the parameters while it is running. Nothing is installed, nothing is uploaded, the whole model lives in one tab.