Data Quality: Validation Frameworks and Anomaly Detection

Data Quality: Ensuring Trustworthy Data at Scale

Data quality is the degree to which data satisfies its intended purpose — accuracy, completeness, consistency, timeliness, and validity.

The Data Quality Dimensions

Core Dimensions:

1. Completeness — are all expected records present?
2. Accuracy — do values reflect reality?
3. Consistency — do related datasets agree?
4. Timeliness — is data available within SLA?
5. Validity — do values conform to business rules?
6. Uniqueness — are there duplicate records?
7. Freshness — how recent is the data?
8. Schema compliance — does the data match expected types and constraints?

Priority by Use Case:

Key Insight: Poor data quality costs organizations an average of $12.9 million annually (Gartner). Data quality is not a one-time check but a continuous engineering discipline embedded throughout the data lifecycle.

Data Quality Framework

Quality Dimensions

Architecture Diagram

Key Concepts

Production Code

Great Expectations Suite

import great_expectations as gx
from great_expectations.core import ExpectationSuite
from great_expectations.checkpoint import SimpleCheckpoint
from great_expectations.dataset import PandasDataset
import pandas as pd
import logging

logger = logging.getLogger(__name__)


class DataQualityValidator:
    """Production data quality validation using Great Expectations."""

    def __init__(self, ge_context_root: str = "great_expectations"):
        self.context = gx.get_context()
        self.suite_name = "production_quality_suite"

    def create_orders_suite(self) -> ExpectationSuite:
        """Create a comprehensive quality suite for orders data."""
        suite = self.context.add_or_update_expectation_suite(
            expectation_suite_name=self.suite_name
        )

        # Schema expectations
        suite.add_expectation(
            gx.expectations.ExpectTableSchemaToMatch(
                schema={
                    "columns": [
                        {"name": "order_id", "type": "string"},
                        {"name": "customer_id", "type": "string"},
                        {"name": "amount", "type": "double"},
                        {"name": "status", "type": "string"},
                        {"name": "created_at", "type": "timestamp"},
                    ]
                }
            )
        )

        # Completeness expectations
        suite.add_expectation(
            gx.expectations.ExpectColumnValuesToNotBeNull(column="order_id")
        )
        suite.add_expectation(
            gx.expectations.ExpectColumnValuesToNotBeNull(column="customer_id")
        )
        suite.add_expectation(
            gx.expectations.ExpectColumnValuesToNotBeNull(column="amount")
        )

        # Uniqueness expectation
        suite.add_expectation(
            gx.expectations.ExpectColumnValuesToBeUnique(column="order_id")
        )

        # Range expectations
        suite.add_expectation(
            gx.expectations.ExpectColumnValuesToBeBetween(
                column="amount",
                min_value=0.01,
                max_value=1000000.0,
            )
        )

        # Set membership expectation
        suite.add_expectation(
            gx.expectations.ExpectColumnValuesToBeInSet(
                column="status",
                value_set=["pending", "processing", "completed", "cancelled", "refunded"],
            )
        )

        # Regex pattern expectation
        suite.add_expectation(
            gx.expectations.ExpectColumnValuesToMatchRegex(
                column="order_id",
                regex=r"^ORD-\d{8}-\d{6}$",
            )
        )

        # Table-level expectations
        suite.add_expectation(
            gx.expectations.ExpectTableRowCountToBeBetween(
                min_value=1000,
                max_value=10000000,
            )
        )

        # Freshness expectation (data should be less than 24 hours old)
        suite.add_expectation(
            gx.expectations.ExpectColumnMaxToBeBetween(
                column="created_at",
                min_value="2024-01-15T00:00:00",
                max_value="2024-01-16T00:00:00",
            )
        )

        return suite

    def validate_batch(self, df: pd.DataFrame) -> dict:
        """
        Validate a batch of data against the quality suite.
        
        Parameters:
            df (pd.DataFrame): Data to validate
            
        Returns:
            dict: Validation results with success status, check counts, and failure details
            
        Example output:
            {
                "success": False,
                "total_checks": 8,
                "passed": 6,
                "failed": 2,
                "failed_expectations": [
                    {"expectation": "expect_column_values_to_not_be_null", "column": "amount", ...},
                    {"expectation": "expect_column_values_to_be_between", "column": "amount", ...}
                ]
            }
        """
        batch = self.context.add_or_update_datasource(
            name="pandas_datasource",
            class_name="PandasDatasource",
        )

        batch_request = batch.build_batch_request(dataframe=df)
        checkpoint_name = "quality_checkpoint"

        checkpoint = self.context.add_or_update_simple_checkpoint(
            name=checkpoint_name,
            validations=[
                {
                    "batch_request": batch_request,
                    "expectation_suite_name": self.suite_name,
                }
            ],
        )

        result = checkpoint.run()
        success = result.success

        # Extract detailed results
        results = {
            "success": success,
            "total_checks": len(result.results),
            "passed": sum(1 for r in result.results if r.success),
            "failed": sum(1 for r in result.results if not r.success),
            "failed_expectations": [
                {
                    "expectation": str(r.expectation_config.expectation_type),
                    "column": r.expectation_config.kwargs.get("column"),
                    "result": r.result,
                }
                for r in result.results
                if not r.success
            ],
        }

        if not success:
            logger.warning(
                f"Data quality validation FAILED: "
                f"{results['failed']}/{results['total_checks']} checks failed"
            )
            for failed in results["failed_expectations"]:
                logger.warning(f"  FAILED: {failed['expectation']} on {failed['column']}")

        return results

Anomaly Detection Pipeline

import numpy as np
import pandas as pd
from scipy import stats
from typing import Dict, List, Tuple
from dataclasses import dataclass
from datetime import datetime, timedelta
import logging

logger = logging.getLogger(__name__)


@dataclass
class AnomalyResult:
    """Result of an anomaly detection check."""
    metric: str
    value: float
    expected_range: Tuple[float, float]
    z_score: float
    is_anomaly: bool
    severity: str  # low, medium, high, critical


class DataQualityAnomalyDetector:
    """Detect anomalies in data quality metrics using statistical methods."""

    def __init__(self, lookback_days: int = 30, z_threshold: float = 2.5):
        self.lookback_days = lookback_days
        self.z_threshold = z_threshold

    def detect_metric_anomalies(
        self,
        current_metrics: Dict[str, float],
        historical_metrics: pd.DataFrame,
    ) -> List[AnomalyResult]:
        """Detect anomalies by comparing current metrics against historical baseline."""
        results = []

        for metric_name, current_value in current_metrics.items():
            if metric_name not in historical_metrics.columns:
                continue

            historical = historical_metrics[metric_name].dropna()
            if len(historical) < 10:
                logger.warning(f"Insufficient history for {metric_name}")
                continue

            mean = historical.mean()
            std = historical.std()

            if std == 0:
                z_score = 0
            else:
                z_score = (current_value - mean) / std

            # Calculate expected range
            expected_low = mean - self.z_threshold * std
            expected_high = mean + self.z_threshold * std

            is_anomaly = abs(z_score) > self.z_threshold

            # Determine severity
            if abs(z_score) > 4:
                severity = "critical"
            elif abs(z_score) > 3:
                severity = "high"
            elif abs(z_score) > self.z_threshold:
                severity = "medium"
            else:
                severity = "low"

            results.append(AnomalyResult(
                metric=metric_name,
                value=current_value,
                expected_range=(expected_low, expected_high),
                z_score=z_score,
                is_anomaly=is_anomaly,
                severity=severity,
            ))

        return results

    def detect_distribution_shift(
        self,
        current_data: pd.Series,
        historical_data: pd.Series,
        metric_name: str,
    ) -> AnomalyResult:
        """Detect distribution shift using Kolmogorov-Smirnov test."""
        # KS test for distribution comparison
        ks_statistic, p_value = stats.ks_2samp(historical_data, current_data)

        is_anomaly = p_value < 0.05  # Significant distribution shift

        return AnomalyResult(
            metric=f"{metric_name}_distribution",
            value=ks_statistic,
            expected_range=(0, 0.1),
            z_score=ks_statistic * 10,  # Scale for consistency
            is_anomaly=is_anomaly,
            severity="high" if is_anomaly else "low",
        )

    def detect_trend_anomaly(
        self,
        time_series: pd.Series,
        metric_name: str,
        window: int = 7,
    ) -> List[AnomalyResult]:
        """Detect anomalies in time-series data using rolling statistics."""
        results = []

        rolling_mean = time_series.rolling(window=window).mean()
        rolling_std = time_series.rolling(window=window).std()

        for i in range(window, len(time_series)):
            if pd.isna(rolling_std.iloc[i]) or rolling_std.iloc[i] == 0:
                continue

            z_score = (time_series.iloc[i] - rolling_mean.iloc[i]) / rolling_std.iloc[i]
            is_anomaly = abs(z_score) > self.z_threshold

            if is_anomaly:
                results.append(AnomalyResult(
                    metric=f"{metric_name}_trend",
                    value=float(time_series.iloc[i]),
                    expected_range=(
                        float(rolling_mean.iloc[i] - self.z_threshold * rolling_std.iloc[i]),
                        float(rolling_mean.iloc[i] + self.z_threshold * rolling_std.iloc[i]),
                    ),
                    z_score=float(z_score),
                    is_anomaly=True,
                    severity="high" if abs(z_score) > 3 else "medium",
                ))

        return results

    def generate_quality_report(
        self,
        df: pd.DataFrame,
        expected_schema: Dict[str, str],
    ) -> Dict:
        """Generate a comprehensive data quality report."""
        report = {
            "timestamp": datetime.utcnow().isoformat(),
            "total_rows": len(df),
            "total_columns": len(df.columns),
            "completeness": {},
            "uniqueness": {},
            "validity": {},
            "anomalies": [],
        }

        # Completeness analysis
        for col in df.columns:
            null_rate = df[col].isnull().mean()
            report["completeness"][col] = {
                "null_rate": float(null_rate),
                "null_count": int(df[col].isnull().sum()),
                "status": "pass" if null_rate < 0.05 else "fail",
            }

        # Uniqueness analysis
        for col in df.columns:
            if "id" in col.lower():
                duplicate_rate = 1 - (len(df[col].unique()) / len(df))
                report["uniqueness"][col] = {
                    "duplicate_rate": float(duplicate_rate),
                    "duplicate_count": int(len(df) - len(df[col].unique())),
                    "status": "pass" if duplicate_rate < 0.01 else "fail",
                }

        # Type validation
        for col, expected_type in expected_schema.items():
            if col in df.columns:
                actual_type = str(df[col].dtype)
                report["validity"][col] = {
                    "expected_type": expected_type,
                    "actual_type": actual_type,
                    "status": "pass" if actual_type == expected_type else "fail",
                }

        return report

Best Practices

1. Implement quality gates at every pipeline stage boundary. Halt the pipeline if quality checks fail.
2. Define data contracts with source system owners specifying schema, quality guarantees, and SLAs.
3. Use Great Expectations or Soda Core for declarative quality rule definitions and automated validation.
4. Monitor anomaly scores over time. Alert when metrics deviate beyond 2.5 standard deviations.
5. Track completeness, freshness, and uniqueness as primary quality KPIs for every pipeline.
6. Implement pre-load validation to reject malformed source data before it enters the pipeline.
7. Build quality dashboards to visualize metrics, trends, and anomalies for stakeholders.
8. Automate remediation for common issues (nulls, duplicates). Create tickets for complex issues.
9. Use statistical methods (Z-score, KS test) for anomaly detection rather than static thresholds.
10. Review quality metrics in weekly data reviews. Update thresholds and add new checks as data patterns evolve.

Data Quality Tool Comparison

Quality Dimension Thresholds

See Also

• 026 - Data Pipeline Testing - Testing quality validation logic
• 024 - Data Ingestion Patterns - Quality at ingestion time
• 027 - Pipeline Monitoring and Observability - Monitoring quality metrics
• 016 - ETL vs ELT - Quality in transformation paradigms
• 025 - Data Quality: Validation Frameworks - This lesson