Data Validation: Ensuring Reliability in Data Pipelines

Data pipelines break. Source systems send unexpected values. Assumptions about data formats turn out to be wrong. Network partitions cause partial writes. The question is not whether your pipeline will encounter bad data, but whether it will detect and handle that bad data gracefully.

Data validation is the practice of checking data against expectations and catching problems before they propagate downstream. Good validation catches errors at the point of entry, contains damage, and alerts operators so problems can be fixed.

Why Validation Matters

Unvalidated data is a liability. Bad data in a data warehouse leads to incorrect reports, wrong business decisions, and eroded trust in data. When stakeholders stop trusting the data, they stop using it and start making decisions based on gut feel.

The cost of fixing bad data increases exponentially the further it travels. A null value caught at ingestion costs minutes to investigate. The same null value in a monthly report that has already been distributed to executives costs days to correct and damages credibility.

Validation is also a form of communication. When you define validation rules, you document your expectations about data quality. This makes implicit requirements explicit and helps source system owners understand what quality standards their data must meet.

flowchart LR
    subgraph Source[Source System]
        S[Raw data
export]
    end
    subgraph SourceLevel[Source-Level Validation]
        SV[Schema check
Type validation
Range check]
    end
    subgraph PipelineLevel[Pipeline Validation]
        PV[Transform rules
Business logic
Cross-record checks]
    end
    subgraph ConsumerLevel[Consumer Validation]
        CV[Referential integrity
Aggregation sanity
Distribution check]
    end
    subgraph Dest[Destination]
        WH[(Data
Warehouse)]
        Q[(Quarantine
folder)]
    end
    S -->|pass| SV
    SV -->|fail| Q
    SV -->|pass| PV
    PV -->|fail| Q
    PV -->|pass| CV
    CV -->|fail| Q
    CV -->|pass| WH

Validation Layers

Effective validation happens at multiple layers of the pipeline.

Source-Level Validation

Validate data at the source before it enters your pipeline. This is often implemented in the source system’s export process or in an early-stage ingestion layer.

def validate_source_record(record, schema):
    """Validate a record against its schema before ingestion."""
    errors = []

    for field, field_type in schema.items():
        if field not in record and not field.get('nullable'):
            errors.append(f"Missing required field: {field}")
        elif field in record:
            value = record[field]
            if not validate_type(value, field_type):
                errors.append(f"Invalid type for {field}: expected {field_type}")

    return errors

def validate_type(value, expected_type):
    """Check if value matches expected type."""
    if expected_type == 'string':
        return isinstance(value, str)
    elif expected_type == 'integer':
        return isinstance(value, int) and not isinstance(value, bool)
    elif expected_type == 'decimal':
        return isinstance(value, (int, float)) and not isinstance(value, bool)
    elif expected_type == 'date':
        return isinstance(value, str) and parse_date(value) is not None
    return True

Pipeline-Level Validation

As data moves through transformations, validate that outputs match expectations. A transformation should not produce NULL values in a column that should never be NULL, or negative values where only positive values make sense.

# Validation checks after a transformation
def validate_transform_output(df, expectations):
    """Validate transformation output against defined expectations."""

    validation_results = {
        'passed': True,
        'errors': [],
        'warnings': []
    }

    # Check for unexpected nulls
    for col in expectations.get('non_nullable', []):
        null_count = df[col].isna().sum()
        if null_count > 0:
            validation_results['passed'] = False
            validation_results['errors'].append(
                f"Unexpected NULLs in {col}: {null_count} rows"
            )

    # Check for value ranges
    for col, (min_val, max_val) in expectations.get('ranges', {}).items():
        out_of_range = ((df[col] < min_val) | (df[col] > max_val)).sum()
        if out_of_range > 0:
            validation_results['passed'] = False
            validation_results['errors'].append(
                f"Values out of range in {col}: {out_of_range} rows"
            )

    # Check for uniqueness
    for col in expectations.get('unique', []):
        duplicate_count = df[col].duplicated().sum()
        if duplicate_count > 0:
            validation_results['warnings'].append(
                f"Duplicate values in {col}: {duplicate_count} rows"
            )

    return validation_results

Consumer-Level Validation

Before data reaches its destination, validate that it meets the quality standards that consumers expect. This is your last chance to catch problems before they affect downstream systems.

-- Validation query before loading into warehouse
SELECT
    'NULL customer_id' AS check_name,
    COUNT(*) AS failure_count
FROM staging_orders
WHERE customer_id IS NULL

UNION ALL

SELECT
    'Negative order amount' AS check_name,
    COUNT(*) AS failure_count
FROM staging_orders
WHERE total_amount < 0

UNION ALL

SELECT
    'Future order date' AS check_name,
    COUNT(*) AS failure_count
FROM staging_orders
WHERE order_date > CURRENT_DATE

UNION ALL

SELECT
    'Missing required product' AS check_name,
    COUNT(*) AS failure_count
FROM staging_orders o
LEFT JOIN dim_product p ON o.product_id = p.product_id
WHERE o.product_id IS NOT NULL AND p.product_id IS NULL;

Schema Validation

Schema validation ensures data conforms to expected structure. This catches structural problems: missing columns, wrong data types, unexpected columns.

Defining Schemas

from pyspark.sql import types as spark_types
from typing import List, Optional

class TableSchema:
    """Define expected schema for a dataset."""

    def __init__(self, name: str):
        self.name = name
        self.fields = []

    def add_string(self, name: str, nullable: bool = False) -> 'TableSchema':
        self.fields.append(spark_types.StructField(
            name, spark_types.StringType(), nullable
        ))
        return self

    def add_integer(self, name: str, nullable: bool = False) -> 'TableSchema':
        self.fields.append(spark_types.StructField(
            name, spark_types.IntegerType(), nullable
        ))
        return self

    def add_decimal(self, name: str, precision: int, scale: int,
                    nullable: bool = False) -> 'TableSchema':
        self.fields.append(spark_types.StructField(
            name, spark_types.DecimalType(precision, scale), nullable
        ))
        return self

    def add_date(self, name: str, nullable: bool = False) -> 'TableSchema':
        self.fields.append(spark_types.StructField(
            name, spark_types.DateType(), nullable
        ))
        return self

    def build(self) -> spark_types.StructType:
        return spark_types.StructType(self.fields)


# Define schema for orders table
orders_schema = (TableSchema('orders')
    .add_integer('order_id', nullable=False)
    .add_integer('customer_id', nullable=False)
    .add_date('order_date', nullable=False)
    .add_decimal('total_amount', 12, 2, nullable=False)
    .build())

Validating Against Schemas

def validate_schema(df, expected_schema):
    """Validate that dataframe matches expected schema."""

    errors = []

    # Check for missing columns
    expected_fields = {f.name for f in expected_schema.fields}
    actual_fields = set(df.columns)
    missing = expected_fields - actual_fields
    if missing:
        errors.append(f"Missing columns: {missing}")

    # Check for extra columns
    extra = actual_fields - expected_fields
    if extra:
        errors.append(f"Unexpected columns: {extra}")

    # Check data types
    for field in expected_schema.fields:
        if field.name in df.columns:
            actual_type = df.schema[field.name].dataType
            if not types_compatible(actual_type, field.dataType):
                errors.append(
                    f"Type mismatch for {field.name}: "
                    f"expected {field.dataType}, got {actual_type}"
                )

    return errors

Business Rule Validation

Beyond schema validation, business rules define what values make sense in context. Schema validation catches structural problems. Business rule validation catches semantic problems.

# Business rule validators
class OrderValidator:
    """Validate business rules for orders."""

    def __init__(self, df):
        self.df = df
        self.errors = []

    def check_positive_amounts(self):
        """Order amounts must be positive."""
        invalid = self.df[self.df['total_amount'] <= 0]
        if len(invalid) > 0:
            self.errors.append(
                f"{len(invalid)} orders with non-positive amounts"
            )
        return self

    def check_valid_customer(self):
        """Customer must exist in customer master."""
        valid_customers = get_valid_customer_ids()
        invalid = self.df[~self.df['customer_id'].isin(valid_customers)]
        if len(invalid) > 0:
            self.errors.append(
                f"{len(invalid)} orders with invalid customer_id"
            )
        return self

    def check_order_date_reasonableness(self):
        """Order date should be within reasonable range."""
        min_date = '2020-01-01'
        max_date = datetime.now().strftime('%Y-%m-%d')
        invalid = self.df[
            (self.df['order_date'] < min_date) |
            (self.df['order_date'] > max_date)
        ]
        if len(invalid) > 0:
            self.errors.append(
                f"{len(invalid)} orders with unreasonable dates"
            )
        return self

    def check_consistent_line_items(self):
        """Sum of line items should equal order total."""
        line_totals = self.df.groupby('order_id')['line_total'].sum()
        order_totals = self.df.groupby('order_id')['total_amount'].first()
        mismatched = line_totals[line_totals != order_totals]
        if len(mismatched) > 0:
            self.errors.append(
                f"{len(mismatched)} orders with line item mismatches"
            )
        return self

    def validate(self):
        """Run all validations and return results."""
        (self
            .check_positive_amounts()
            .check_valid_customer()
            .check_order_date_reasonableness()
            .check_consistent_line_items())

        return {
            'passed': len(self.errors) == 0,
            'errors': self.errors
        }

Anomaly Detection

Beyond rule-based validation, anomaly detection uses statistical methods to identify unusual patterns that might indicate problems.

from scipy import stats
import numpy as np

def detect_statistical_anomalies(series, z_threshold=3.0):
    """Detect anomalies using z-score method."""

    z_scores = np.abs(stats.zscore(series.dropna()))
    anomaly_mask = z_scores > z_threshold

    return {
        'anomaly_count': anomaly_mask.sum(),
        'anomaly_indices': np.where(anomaly_mask)[0],
        'anomaly_values': series[anomaly_mask].tolist()
    }

def detect_distribution_shift(old_values, new_values, threshold=0.1):
    """Detect if new data has shifted significantly from historical data."""

    # Compare mean
    mean_shift = abs(new_values.mean() - old_values.mean()) / old_values.mean()

    # Compare std dev
    std_shift = abs(new_values.std() - old_values.std()) / old_values.std()

    return {
        'mean_shift_detected': mean_shift > threshold,
        'std_shift_detected': std_shift > threshold,
        'mean_shift_pct': mean_shift * 100,
        'std_shift_pct': std_shift * 100
    }

def check_for_null_pattern_anomalies(df):
    """Check if null patterns have changed compared to historical baseline."""

    # Compute current null pattern
    current_nulls = df.isnull().mean()

    # Compare to historical baseline
    historical_nulls = load_historical_null_patterns(df.columns)

    shifts = {}
    for col in df.columns:
        if col in historical_nulls:
            shift = abs(current_nulls[col] - historical_nulls[col])
            if shift > 0.05:  # 5% threshold
                shifts[col] = {
                    'current_null_rate': current_nulls[col],
                    'historical_null_rate': historical_nulls[col],
                    'shift': shift
                }

    return shifts

Handling Validation Failures

What happens when validation fails? The answer depends on the severity and your pipeline design.

Quarantine and Alert

def handle_validation_failure(validation_result, record_batch, context):
    """Handle validation failures."""

    if validation_result['severity'] == 'CRITICAL':
        # Quarantine bad records
        quarantine_batch(record_batch, context)

        # Alert immediately
        alert_operations(
            alert_type='DATA_QUALITY_CRITICAL',
            dataset=context.dataset_name,
            error_count=len(validation_result['errors']),
            errors=validation_result['errors']
        )

        # Block pipeline continuation
        raise ValidationException(
            f"Critical validation failure: {validation_result['errors']}"
        )

    elif validation_result['severity'] == 'WARNING':
        # Log warning but continue
        log_warning(
            f"Validation warning: {validation_result['errors']}"
        )

        # Track for monitoring
        record_validation_warning(context.dataset_name, validation_result)

    elif validation_result['severity'] == 'INFO':
        # Just log for auditing
        log_validation_info(context.dataset_name, validation_result)

Data Quarantine

def quarantine_batch(batch, context):
    """Move failed records to quarantine for investigation."""

    quarantine_path = (
        f"s3://data-lake/quarantine/"
        f"{context.dataset_name}/"
        f"{datetime.now().strftime('%Y%m%d/%H%M%S')}/"
    )

    # Write failed records with metadata
    failed_df = batch.filter(batch._validation_failed)
    failed_df.write.parquet(quarantine_path)

    # Write metadata about the failure
    metadata = {
        'dataset': context.dataset_name,
        'validation_errors': context.validation_errors,
        'record_count': failed_df.count(),
        'quarantine_time': datetime.now().isoformat()
    }
    write_metadata(quarantine_path, metadata)

    return quarantine_path

Validation at Scale

Validating large datasets requires careful design to avoid becoming a bottleneck.

Sampling-Based Validation

For very large datasets, validate a sample rather than every record.

def sample_validation(df, sample_size=10000, confidence_level=0.95):
    """Validate a statistical sample instead of full dataset."""

    if len(df) <= sample_size:
        return validate_full_dataset(df)

    # Stratified sampling to ensure representation
    sample = df.sample(n=sample_size)

    validation_result = validate_full_dataset(sample)

    # Scale error estimates to full dataset
    scale_factor = len(df) / sample_size
    estimated_total_errors = len(validation_result['errors']) * scale_factor

    return {
        'validation_type': 'sampled',
        'sample_size': sample_size,
        'total_records': len(df),
        'errors_found': validation_result['errors'],
        'estimated_total_errors': estimated_total_errors,
        'confidence_level': confidence_level
    }

Distributed Validation

When validating across a distributed dataset, run validation in parallel.

def distributed_validation(df, validation_functions, partition_count=100):
    """Run validations in parallel across dataframe partitions."""

    # Repartition for parallelism
    df = df.repartition(partition_count)

    # Broadcast validation functions to all executors
    broadcast_validators = spark.sparkContext.broadcast(validation_functions)

    def validate_partition(partition):
        """Validate a single partition."""
        validator = broadcast_validators.value
        errors = []
        for record in partition:
            for check_fn in validator:
                result = check_fn(record)
                if not result['passed']:
                    errors.append(result)
        return errors

    # Map validation across partitions
    error_rdd = df.rdd.mapPartitions(validate_partition)
    all_errors = error_rDD.collect()

    return {
        'total_errors': len(all_errors),
        'errors_by_type': group_errors_by_type(all_errors),
        'partition_error_distribution': get_partition_distribution(error_rdd)
    }

Building Validation into Pipelines

Validation should be a first-class citizen in your pipeline, not an afterthought.

from data_pipeline import Pipeline, Stage

class ValidationStage(Stage):
    """Pipeline stage for data validation."""

    def __init__(self, validators, failure_mode='quarantine'):
        self.validators = validators
        self.failure_mode = failure_mode

    def process(self, df):
        for validator in self.validators:
            result = validator.validate(df)

            if not result['passed']:
                if self.failure_mode == 'quarantine':
                    self._handle_quarantine(df, result)
                elif self.failure_mode == 'drop':
                    df = self._drop_invalid_records(df, validator)
                elif self.failure_mode == 'fail':
                    raise ValidationException(result['errors'])

        return df

# Build pipeline with validation
pipeline = (
    Pipeline()
    .stage(IngestionStage())
    .stage(ValidationStage([
        OrderValidator(),
        CustomerValidator(),
        ProductValidator()
    ], failure_mode='quarantine'))
    .stage(TransformStage())
    .stage(LoadStage())
    .build()
)

Metrics and Monitoring

Track validation metrics over time to identify trends and regressions.

# Metrics to track
validation_metrics = {
    'records_validated': Counter('data_validation_records_total'),
    'validation_failures': Counter('data_validation_failures_total'),
    'validation_errors': Histogram('data_validation_error_rate'),
    'quarantine_records': Counter('data_quarantine_records_total'),
    'validation_latency': Histogram('data_validation_duration_seconds')
}

# Log validation results for monitoring
def log_validation_metrics(result, context):
    """Emit metrics for monitoring."""

    validation_metrics['records_validated'].inc(context.record_count)
    validation_metrics['validation_errors'].observe(
        result['error_rate']
    )

    if result['failure_count'] > 0:
        validation_metrics['validation_failures'].labels(
            dataset=context.dataset_name,
            error_type=result['error_type']
        ).inc(result['failure_count'])

Data Validation Trade-Offs

Aspect	Strict Validation	Lenient Validation	Ingestion-Time	Query-Time
Data quality	High (blocks bad data)	Medium (allows some noise)	High (fail fast)	Medium (catch drift)
Pipeline reliability	Risk of blocking	Low risk of blocking	Risk at load	No load risk
Latency	Immediate feedback	Deferred feedback	Immediate	On query
Catch edge cases	Schema violations, type errors	Logic errors, anomalies	Structural issues	Distribution shifts
Operational burden	High (fix failures fast)	Low	High	Medium
Best for	Financial data, regulated	Log data, exploratory	Real-time pipelines	Warehouse analytics

Data Validation Production Failure Scenarios

Strict validation blocking a critical pipeline

A pipeline uses strict validation at ingestion. An upstream source system pushes a batch with a new enum value that was added without notice. Validation rejects the entire batch. The pipeline does not run, downstream reports are stale, and the on-call engineer gets paged at 2am.

This failure mode is insidious because the pipeline was working correctly—it did exactly what it was designed to do. The problem is that the validation rules were written against a snapshot of the source system’s schema at a point in time, not against the actual current state of the source. When the upstream team added a new enum value, they had no obligation to notify downstream consumers, and nobody had set up a channel for that communication anyway. The first time anyone knew about the change was when the batch arrived and got rejected.

The 2am page is a symptom of a deeper design problem: treating all validation failures as equally critical. A new enum value in a field that has no downstream impact is fundamentally different from a missing primary key that will cause join failures across your entire warehouse. Conflating these makes your pipeline brittle without making it safer.

Mitigation: Implement validation severity levels. CRITICAL blocks the pipeline; WARNING quarantines but continues; INFO logs. New enum values should produce WARNING (quarantine the new value for review) not CRITICAL (block everything). Define a review SLA for quarantine contents—validation failures that sit in quarantine for more than 48 hours should escalate automatically. Create a shared enum registry that source systems can update proactively, so validation rules evolve alongside the source schema rather than being frozen in place.

False positive validation alerts causing alert fatigue

A validation rule checks that customer_region IN ('Northeast', 'Southwest', 'Northwest', 'Southeast'). A new region ‘Midwest’ is added to the business but the validation rule is never updated. Every subsequent batch triggers alerts. After a week, the team silences the alert. A real data quality problem two days later goes undetected.

The danger here is not the silence itself—it is what the silence obscures. When a rule fires constantly for a week and nobody fixes it, the team has already decided, implicitly, that this category of alert does not matter. That decision becomes organizational muscle memory. The next time the rule fires for a real problem, the same reflex applies. Alert fatigue is not just about noise; it reshapes how teams respond to signal.

The rule was also a maintenance liability from the start. Hardcoding a fixed set of region values means the rule itself must be updated every time the business adds or reorganizes regions. This is a tax on every new region addition—a cost that nobody explicitly agreed to pay. Over time, the rule becomes a static snapshot of a dynamic business, and every mismatch between snapshot and reality generates a spurious alert.

The real failure happened two days after silencing, when a genuine data quality problem emerged. Because the alert channel had been rendered untrustworthy, nobody was watching it. The business user who eventually found the problem had no idea how long it had been there or how many reports had been built on bad data in the interim.

Mitigation: Track alert frequency per rule. Rules that fire more than once per week deserve a review—are they testing for real errors or outdated assumptions? Archive rules that are no longer relevant instead of leaving them to generate noise. When archiving a rule, document why it was created and when it became obsolete. This creates an audit trail that helps future engineers understand the evolution of your validation logic without having to reverse-engineer decisions from stale code.

Quarantine folder growing unbounded

The quarantine folder has no retention policy. Every validation failure writes files indefinitely. After a year, the quarantine folder is 500GB with millions of small Parquet files. Reading the quarantine folder listing takes minutes, and storage costs are uncontrolled.

The problem usually starts innocently. Someone writes a validation rule that occasionally fails on edge cases. The failures are real, so nobody questions quarantine-ingest the records. But the rule is not fixed because the edge cases are “benign”—or so the team believes at the time. The quarantine folder grows quietly. Months pass. The team that built the pipeline moves on. The new team does not even know the quarantine folder exists. When someone finally notices, the folder is already measured in terabytes and the metadata about what each file represents has long since drifted out of sync with reality.

The operational impact is not just storage cost. A massive quarantine folder means that any investigation into validation failures requires scanning millions of files before finding the relevant ones. Engineers stop investigating quarantine contents because the friction is too high. The quarantine folder stops serving its purpose—it becomes a graveyard where bad data goes to be forgotten, not reviewed.

Small Parquet files are a compounding problem. Each validation failure generates its own small Parquet file. Parquet is a columnar format that compresses well for large files but has significant per-file overhead. Millions of small files do not compress efficiently, and listing them puts pressure on the filesystem metadata layer. In cloud storage environments, this can translate to significant egress and API call costs on top of raw storage.

Mitigation: Set TTL on quarantine files (7 days is typical for investigation). Compress quarantine files before archival. Alert when quarantine folder growth exceeds expected baseline. Implement a quarantine budget per dataset—if the quarantine folder for a given pipeline grows beyond a defined threshold, automatic compression and archival triggers. Treat quarantine folder size as a pipeline health metric, not an archival concern.

Statistical anomaly detection on sparse data

An anomaly detection check runs on a dataset with 100 rows and flags a value as anomalous because it is 3 standard deviations from the mean. The dataset is too small for statistical anomaly detection—the standard deviation is meaningless with so few data points.

The issue is not that the algorithm is wrong; it is that the algorithm is being applied outside the conditions where its outputs are meaningful. Standard deviation is a measure of spread, and with 100 points, a single outlier can swing the standard deviation dramatically in ways that do not reflect the underlying distribution. With 100 rows, even a moderately unusual value—something that is genuinely just a legitimate part of the distribution—can appear to be 3 standard deviations away simply because the sample is too small to estimate the true population parameters accurately.

This is especially problematic for data that has natural skew. Transaction amounts, user session durations, and API response times are often log-normally distributed. A dataset of 100 transaction amounts might look like it has outliers by z-score, when in reality those values are entirely consistent with a log-normal distribution that just has not had enough samples to look smooth. The statistical check is not wrong in a mathematical sense—it is measuring exactly what it is supposed to measure—but the interpretation is wrong because the preconditions for that interpretation are not met.

There is also a detection problem: with small datasets, the opposite failure mode is equally likely. A real anomaly can hide in plain sight. If a dataset normally has 10,000 rows and today it has 100, the anomaly detection will be completely unreliable. The check might miss a real problem because the baseline is wrong.

Mitigation: Set minimum dataset size for statistical checks (e.g., require at least 1,000 records). Use rule-based validation for small datasets and statistical methods only when sample sizes are meaningful. When sample sizes are borderline, use non-parametric methods—median absolute deviation (MAD) is more robust to outliers than standard deviation for small samples. Log the effective sample size alongside anomaly detection results so that investigators know how much confidence to place in the flagged values.

Data Validation Anti-Patterns

Validating everything at ingestion. Putting all validation logic at the entry point creates a bottleneck and makes debugging hard when failures occur. Spread validation across layers so each layer catches what it handles best—ingestion catches structural issues, pipeline catches logic errors, consumer catches downstream impact.

No validation at all. Trusting source data because “they promised it is clean.” Source systems change. Assumptions break. Every pipeline needs validation regardless of how trustworthy the source seems today.

Keeping quarantine forever. Quarantine files accumulating forever because “we might need them later.” Without TTL and review processes, quarantine becomes a data swamp that consumes storage and provides no value. Set retention policies and review quotas.

Silent warnings. Validation failures that log warnings but continue without alerting. Silent failures mean nobody knows about data quality problems until a business user notices and reports it. WARN-level validation failures should still emit metrics and periodically alert if their rate exceeds threshold.

Data Validation Quick Recap

• Validate at multiple layers: source catches structural issues, pipeline catches logic errors, consumer catches downstream impact.
• Use severity levels (CRITICAL/WARNING/INFO) to distinguish blocking failures from quarantined warnings.
• Statistical anomaly detection requires sufficient sample sizes—do not apply z-score methods to datasets under ~1,000 rows.
• Quarantine requires TTL policies and regular review; unbounded quarantine is a data swamp.
• Alert fatigue kills data quality programs: prune outdated rules and track alert frequency per validation rule.

For related reading on data quality, see Data Governance for the broader framework of data quality management, or Audit Trails for how validation fits into overall data auditability.

Data validation is not optional. Every pipeline encounters bad data eventually. The difference between a resilient pipeline and a problematic one is whether validation catches issues early.

The key principles:

• Validate at multiple layers: source, pipeline, and consumer
• Define schemas and enforce them programmatically
• Implement business rules beyond structural validation
• Use statistical anomaly detection for unexpected patterns
• Handle failures gracefully with quarantine and alerting
• Track validation metrics over time

For related reading on data quality, see Data Governance for the broader framework of data quality management, or Audit Trails for how validation fits into overall data auditability.