Data validation

Data Validation Patterns

Pandera Schema Validation

import pandera as pa
from pandera import Column, DataFrameSchema, Check

schema = DataFrameSchema({
    "user_id": Column(str, nullable=False, unique=True),
    "amount": Column(float, Check.greater_than_or_equal_to(0), nullable=False),
    "category": Column(str, Check.isin(["A", "B", "C"])),
    "ts": Column(pa.DateTime, nullable=False),
    "score": Column(float, [Check.in_range(0.0, 1.0)], nullable=True),
})

# Validate and get informative errors
try:
    validated_df = schema.validate(df, lazy=True)
except pa.errors.SchemaErrors as e:
    print(e.failure_cases)  # DataFrame with row-level failures

Pandera with Pydantic-style Models

from pandera.typing import DataFrame, Series
import pandera as pa

class TransactionSchema(pa.DataFrameModel):
    user_id: Series[str] = pa.Field(nullable=False)
    amount: Series[float] = pa.Field(ge=0)
    category: Series[str] = pa.Field(isin=["A", "B", "C"])

    class Config:
        coerce = True
        strict = True

@pa.check_types
def process_transactions(df: DataFrame[TransactionSchema]) -> DataFrame[TransactionSchema]:
    return df.assign(amount_usd=df["amount"] / 100)

Great Expectations Suite

import great_expectations as gx

context = gx.get_context()
datasource = context.sources.add_pandas_filesystem(
    name="local_parquet",
    base_directory="data/",
)
asset = datasource.add_parquet_asset("features")
batch = asset.build_batch_request()

suite = context.add_or_update_expectation_suite("feature_quality")
validator = context.get_validator(batch_request=batch, expectation_suite_name="feature_quality")

validator.expect_column_values_to_not_be_null("user_id")
validator.expect_column_values_to_be_between("amount", min_value=0, max_value=100_000)
validator.expect_column_proportion_of_unique_values_to_be_between("user_id", min_value=0.95)
validator.expect_column_mean_to_be_between("score", min_value=0.3, max_value=0.7)
validator.save_expectation_suite(discard_failed_expectations=False)

checkpoint = context.add_or_update_checkpoint(
    name="daily_checkpoint",
    validator=validator,
)
result = checkpoint.run()
if not result["success"]:
    raise RuntimeError("Data quality check failed")

Statistical Drift Detection

from scipy import stats
import numpy as np

def ks_drift(ref: np.ndarray, curr: np.ndarray, threshold: float = 0.05) -> dict:
    stat, p_value = stats.ks_2samp(ref, curr)
    return {"statistic": stat, "p_value": p_value, "drift": p_value < threshold}

def psi(expected: np.ndarray, actual: np.ndarray, n_bins: int = 10) -> float:
    """Population Stability Index. PSI > 0.25 = significant shift."""
    breaks = np.percentile(expected, np.linspace(0, 100, n_bins + 1))
    breaks[0], breaks[-1] = -np.inf, np.inf
    exp_pct = np.histogram(expected, breaks)[0] / len(expected)
    act_pct = np.histogram(actual, breaks)[0] / len(actual)
    exp_pct = np.where(exp_pct == 0, 1e-4, exp_pct)
    act_pct = np.where(act_pct == 0, 1e-4, act_pct)
    return float(np.sum((act_pct - exp_pct) * np.log(act_pct / exp_pct)))

Evidently Report

from evidently.report import Report
from evidently.metric_preset import DataDriftPreset, DataQualityPreset

report = Report(metrics=[DataDriftPreset(), DataQualityPreset()])
report.run(reference_data=ref_df, current_data=curr_df)
report.save_html("drift_report.html")

# Programmatic access
result = report.as_dict()
drift_detected = result["metrics"][0]["result"]["dataset_drift"]

Data Contract (YAML-based)

# contracts/features.yaml
version: "1.0"
dataset: user_features
columns:
  user_id: {type: string, nullable: false, unique: true}
  daily_spend: {type: float, min: 0, max: 100000}
  tx_count: {type: integer, min: 0}
  score: {type: float, min: 0.0, max: 1.0, nullable: true}
freshness:
  max_age_hours: 25
row_count:
  min: 10000