Data Validation

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

| Dimension | Description | Example Check | |-----------|-------------|---------------| | Completeness | No missing required values | null_count(col) == 0 | | Uniqueness | No unexpected duplicates | unique_count(id) == row_count | | Consistency | Values conform to rules | min(age) >= 0 | | Accuracy | Values are correct | mean(price) within expected range | | Timeliness | Data is fresh enough | max(timestamp) > now() - 1h | | Validity | Values in expected domain | col IN ('A', 'B', 'C') |

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 as pa
from pandera 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  {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 - Comprehensive data validation with GX and Pandera
• scripts/data_contract.py - Data contract enforcement engine

References

See [references/REFERENCE.md](references/REFERENCE.md) for tool comparisons and patterns.

Source & license

This open-source skill is cataloged on AgentStack and links to its original source — we do not rehost the code.

• Author: timwukp
• Source: timwukp/MLOps-agent-skills
• License: Apache-2.0

Install and usage instructions live in the source repository linked above.