Data Quality Monitoring

Data Quality Monitoring

Trust your data, or fix it before anyone else sees it.

Quality Dimensions

Anomaly Detection

Row Count Monitoring (Z-Score)

def check_row_count(source: str, current_count: int,
                    expected: float, stddev: float,
                    threshold_pct: float = 0.25) -> dict:
    """Detect row count anomalies using rolling baseline."""
    z_score = abs(current_count - expected) / max(stddev, 1)
    deviation_pct = abs(current_count - expected) / max(expected, 1)

    return {
        "source": source,
        "current": current_count,
        "expected": round(expected),
        "deviation_pct": round(deviation_pct, 4),
        "z_score": round(z_score, 2),
        "is_anomaly": deviation_pct > threshold_pct,
        "severity": _severity(deviation_pct),
    }

def _severity(deviation: float) -> str:
    if deviation > 0.5: return "CRITICAL"
    if deviation > 0.25: return "HIGH"
    if deviation > 0.1: return "MEDIUM"
    return "LOW"

Severity Actions

Schema Drift Detection

def detect_schema_drift(actual_cols: dict, expected_cols: dict,
                        strict: bool = False) -> dict:
    """Compare actual vs expected schema."""
    missing = set(expected_cols) - set(actual_cols)
    added = set(actual_cols) - set(expected_cols)
    type_changes = {
        col: (expected_cols[col], actual_cols[col])
        for col in set(actual_cols) & set(expected_cols)
        if expected_cols[col] != actual_cols[col]
    }

    return {
        "has_drift": bool(missing or (strict and added) or type_changes),
        "missing_columns": list(missing),
        "added_columns": list(added),
        "type_changes": type_changes,
        "severity": "HIGH" if missing or type_changes else "LOW",
    }

Null Ratio Monitoring

def check_null_ratios(columns: dict, total_rows: int,
                      thresholds: dict = None) -> list:
    """Flag columns exceeding null thresholds."""
    default_threshold = 0.1
    thresholds = thresholds or {}
    issues = []

    for col, null_count in columns.items():
        ratio = null_count / max(total_rows, 1)
        threshold = thresholds.get(col, default_threshold)
        if ratio > threshold:
            issues.append({
                "column": col,
                "null_count": null_count,
                "null_ratio": round(ratio, 4),
                "threshold": threshold,
                "severity": "HIGH" if ratio > threshold * 2 else "MEDIUM",
            })

    return issues

Freshness Check

from datetime import datetime

def check_freshness(latest_timestamp: datetime,
                    max_age_hours: int = 24) -> dict:
    """Check data freshness against maximum allowed age."""
    age_hours = (datetime.now() - latest_timestamp).total_seconds() / 3600

    return {
        "fresh": age_hours <= max_age_hours,
        "latest_record": latest_timestamp.isoformat(),
        "age_hours": round(age_hours, 2),
        "max_age_hours": max_age_hours,
        "status": "FRESH" if age_hours <= max_age_hours else "STALE",
    }

Data Protection Patterns

Safe Write (Prevent Empty Overwrites)

def safe_write(current_count: int, existing_count: int,
               min_rows: int = 100, max_drop_pct: float = 0.5) -> list:
    """Validate before overwriting production data."""
    failures = []

    if current_count == 0 and existing_count > 0:
        failures.append(f"NO_EMPTY_OVERWRITE: cannot replace {existing_count} rows with 0")

    if existing_count > 0 and current_count < existing_count * (1 - max_drop_pct):
        drop = (1 - current_count / existing_count) * 100
        failures.append(f"MAX_DROP: {drop:.1f}% exceeds {max_drop_pct * 100}% threshold")

    if current_count < min_rows:
        failures.append(f"MIN_ROWS: {current_count} below minimum {min_rows}")

    return failures  # Empty list = safe to write

Protection Decision Matrix

Quality Report

def generate_quality_report(source: str, row_count: int,
                            checks: dict) -> dict:
    """Combine all checks into a single report."""
    report = {
        "source": source,
        "timestamp": datetime.now().isoformat(),
        "row_count": row_count,
        "checks": checks,
        "overall_status": "PASS" if all(
            not c.get("is_anomaly") and c.get("severity", "LOW") != "HIGH"
            for c in checks.values() if isinstance(c, dict)
        ) else "FAIL",
    }
    return report

Baseline Management

Rolling Baseline Update

def update_baseline(history: list, new_value: int,
                    window_size: int = 30) -> dict:
    """Update rolling baseline with new observation."""
    history = history[-window_size:] + [new_value]
    avg = sum(history) / len(history)
    variance = sum((x - avg) ** 2 for x in history) / len(history)

    return {
        "avg": round(avg, 2),
        "stddev": round(variance ** 0.5, 2),
        "samples": len(history),
        "last_value": new_value,
        "history": history,
    }

Implementation Checklist

• Row count anomaly detection on every pipeline run
• Schema drift check against registered schema
• Null ratio thresholds per column documented
• Freshness alerts for time-sensitive data
• Safe-write guards on production tables
• Rolling baselines updated after successful runs
• Quality reports stored for trend analysis
• Alerting integrated (email, Slack, PagerDuty)