Imported from ehadziabdic/WAgents (skills/data-validation/SKILL.md). Install upstream with npx skills add ehadziabdic/WAgents --skill data-validation. Copyright stays with the author (Apache-2.0).

Data Validation for ML Pipelines

Overview

Data validation ensures that data feeding ML models meets quality, schema, and business requirements. Poor data quality is the #1 cause of ML model failures in production.

When to Use This Skill

• Adding quality gates to data pipelines
• Profiling new datasets before model training
• Setting up data contracts between teams
• Debugging unexpected model performance degradation
• Automating data quality checks in CI/CD

Data Quality Dimensions

Step-by-Step Instructions

1. Great Expectations Validation

import great_expectations as gx

# Initialize context
context = gx.get_context()

# Connect to data
datasource = context.data_sources.add_pandas("my_datasource")
asset = datasource.add_dataframe_asset("my_asset")
batch = asset.add_batch_definition_whole_dataframe("my_batch").get_batch(
    batch_parameters={"dataframe": df}
)

# Create expectations
suite = context.suites.add(gx.ExpectationSuite(name="feature_quality"))
suite.add_expectation(gx.expectations.ExpectColumnValuesToNotBeNull(column="user_id"))
suite.add_expectation(gx.expectations.ExpectColumnValuesToBeBetween(
    column="age", min_value=0, max_value=150
))
suite.add_expectation(gx.expectations.ExpectColumnValuesToBeUnique(column="user_id"))
suite.add_expectation(gx.expectations.ExpectTableRowCountToBeBetween(
    min_value=1000, max_value=10000000
))

# Validate
results = batch.validate(suite)
if not results.success:
    for result in results.results:
        if not result.success:
            print(f"FAILED: {result.expectation_config}")

2. Pandera Schema Validation

import pandera.pandas as pa  # pandas-specific namespace (pandera 0.20+)
from pandera.pandas import Column, Check, DataFrameSchema

# Define schema
schema = DataFrameSchema({
    "user_id": Column(int, Check.gt(0), unique=True, nullable=False),
    "age": Column(int, Check.in_range(0, 150), nullable=False),
    "income": Column(float, Check.gt(0), nullable=True),
    "category": Column(str, Check.isin(["A", "B", "C"]), nullable=False),
    "score": Column(float, Check.in_range(0.0, 1.0)),
    "created_at": Column(pa.DateTime, nullable=False),
}, coerce=True)

# Validate
validated_df = schema.validate(df, lazy=True)  # lazy=True collects all errors

# Decorator-based validation
@pa.check_input(schema)
def train_model(df):
    """Input data is automatically validated."""
    pass

3. Data Profiling

# Using ydata-profiling (formerly pandas-profiling)
from ydata_profiling import ProfileReport

profile = ProfileReport(df, title="Training Data Profile", minimal=True)
profile.to_file("data_profile.html")

# Using whylogs
import whylogs as why

results = why.log(df)
profile = results.profile()
view = profile.view()

# Get summary statistics
summary = view.to_pandas()
print(summary[["distribution/mean", "distribution/stddev", "types/fractional"]])

4. Data Contracts

# data_contract.yaml
contract:
  name: user_features_v2
  owner: ml-team
  description: User feature table for recommendation model
  sla:
    freshness: 1h
    completeness: 99.5%
    row_count_min: 100000

  schema:
    columns:
      - name: user_id
        type: integer
        nullable: false
        unique: true
      - name: age
        type: integer
        nullable: false
        checks:
          - min: 0
          - max: 150
      - name: lifetime_value
        type: float
        nullable: true
        checks:
          - min: 0.0

  quality_rules:
    - rule: "duplicate_ratio < 0.001"
    - rule: "null_ratio(income) < 0.05"
    - rule: "mean(age) BETWEEN 25 AND 55"

import yaml

def enforce_contract(df, contract_path):
    """Enforce a data contract against a DataFrame."""
    with open(contract_path) as f:
        contract = yaml.safe_load(f)["contract"]

    violations = []

    # Check SLA
    if "sla" in contract:
        sla = contract["sla"]
        if "row_count_min" in sla and len(df) < sla["row_count_min"]:
            violations.append(f"Row count {len(df)} < min {sla['row_count_min']}")

    # Check schema
    for col_spec in contract["schema"]["columns"]:
        col = col_spec["name"]
        if col not in df.columns:
            violations.append(f"Missing column: {col}")
            continue
        if not col_spec.get("nullable", True) and df[col].isnull().any():
            violations.append(f"Null values in non-nullable column: {col}")

    return violations

5. Automated Data Testing in CI/CD

# tests/test_data_quality.py
import pytest
import pandera.pandas as pa

@pytest.fixture
def training_data():
    return pd.read_parquet("data/training.parquet")

def test_no_nulls_in_target(training_data):
    assert training_data["target"].isnull().sum() == 0

def test_feature_ranges(training_data):
    assert training_data["age"].between(0, 150).all()
    assert training_data["score"].between(0, 1).all()

def test_no_duplicates(training_data):
    assert training_data["user_id"].is_unique

def test_data_freshness(training_data):
    max_ts = training_data["updated_at"].max()
    assert (pd.Timestamp.now() - max_ts).total_seconds() < 3600

def test_distribution_stability(training_data, reference_data):
    """Check that feature distributions haven't shifted dramatically."""
    for col in ["age", "income", "score"]:
        ref_mean = reference_data[col].mean()
        cur_mean = training_data[col].mean()
        assert abs(cur_mean - ref_mean) / ref_mean < 0.2  # 20% tolerance

6. Schema Evolution Handling

def validate_schema_evolution(current_schema, new_schema):
    """Check if schema change is backward compatible."""
    breaking_changes = []

    current_cols = {c.name: c for c in current_schema}
    new_cols = {c.name: c for c in new_schema}

    # Removed columns are breaking
    for col in current_cols:
        if col not in new_cols:
            breaking_changes.append(f"Column removed: {col}")

    # Type changes are breaking
    for col in current_cols:
        if col in new_cols and current_cols[col].type != new_cols[col].type:
            breaking_changes.append(
                f"Type changed for {col}: {current_cols[col].type} -> {new_cols[col].type}"
            )

    # New non-nullable columns are breaking
    for col in new_cols:
        if col not in current_cols and not new_cols[col].nullable:
            breaking_changes.append(f"New non-nullable column: {col}")

    return breaking_changes

Best Practices

1. Validate early - Check data at ingestion, not just before training
2. Use data contracts - Formal agreements between data producers and consumers
3. Profile regularly - Track statistical properties over time to detect drift
4. Automate in CI/CD - Run data tests alongside code tests
5. Alert, don't just log - Failed validations should trigger notifications
6. Version your expectations - Track validation rules in version control
7. Start simple - Begin with null checks and type validation, add complexity later
8. Test edge cases - Empty datasets, single-row datasets, extreme values
9. Document exceptions - When you skip validation, document why
10. Monitor validation pass rates - Track the trend of validation failures

Scripts

• scripts/validate_data.py - Dependency-light data validation (pandas-based checks, profiling, YAML schema)
• scripts/data_contract.py - Data contract enforcement engine

References

See references/REFERENCE.md for tool comparisons and patterns.

Related skills

Upstream: data-ingestion (landed raw data) · Downstream: feature-engineering (validated, contract-conformant data) See also: ml-testing for wiring data checks into CI gates · model-drift-detection for the production-time counterpart of these checks