Data Validation Helper

Data Validation Helper

Workflow en 8 étapes

1. Inventaire et règles métier

• Lister chaque colonne : type attendu, nullabilité, plage (min/max), format (regex, pattern date), unicité, enum de valeurs autorisées.
• Documenter dans un fichier versionné (expectations.yaml, schema.yaml ou conftest.py).
• Critère clé : distinguer règle bloquante (type incorrect, PK dupliquée) de warning tolerable (outlier statistique, valeur rare).

2. Schema validation — fail fast

JSON Schema (API REST) :

from jsonschema import validate, ValidationError

schema = {
    "type": "object",
    "properties": {
        "user_id": {"type": "integer"},
        "email":   {"type": "string", "format": "email"},
        "amount":  {"type": "number",  "minimum": 0}
    },
    "required": ["user_id", "email", "amount"]
}

try:
    validate(instance=payload, schema=schema)
except ValidationError as e:
    raise ValueError(f"Payload invalide : {e.message}")

Pydantic v2 (Python objects) :

from pydantic import BaseModel, field_validator
from decimal import Decimal

class Transaction(BaseModel):
    user_id: int
    amount: Decimal
    currency: str

    @field_validator("currency")
    @classmethod
    def check_currency(cls, v: str) -> str:
        if v not in {"TND", "EUR", "USD"}:
            raise ValueError(f"Devise inconnue : {v}")
        return v

Pandera (DataFrames) :

import pandera as pa

schema = pa.DataFrameSchema({
    "user_id": pa.Column(int,  nullable=False),
    "amount":  pa.Column(float, pa.Check.ge(0)),
    "status":  pa.Column(str,  pa.Check.isin(["pending","done","failed"]))
})

validated_df = schema.validate(df, lazy=True)  # lazy=True collecte toutes les erreurs

3. Dimensions de qualité à mesurer

4. Framework — choix selon contexte

dbt schema.yml :

models:
  - name: transactions
    columns:
      - name: id
        tests: [not_null, unique]
      - name: status
        tests:
          - accepted_values:
              values: ["pending", "done", "failed"]
      - name: user_id
        tests:
          - relationships:
              to: ref('users')
              field: id

Great Expectations — checkpoint minimal :

import great_expectations as gx

context = gx.get_context()
batch = context.sources.pandas_default.read_dataframe(df)
suite  = context.add_expectation_suite("transactions_suite")

batch.expect_column_values_to_not_be_null("user_id")
batch.expect_column_values_to_be_between("amount", min_value=0, max_value=1_000_000)
batch.expect_column_values_to_be_in_set("status", ["pending","done","failed"])

result = context.run_checkpoint(checkpoint_name="daily_check")
if not result["success"]:
    raise RuntimeError("Checkpoint GE échoué — données en quarantaine.")

5. Tests statistiques et détection de dérive

from scipy.stats import ks_2samp
import numpy as np

# Comparer la distribution actuelle vs référence (baseline)
stat, p_value = ks_2samp(baseline_amounts, current_amounts)
if p_value < 0.05:
    alert(f"Dérive détectée sur 'amount' (KS p={p_value:.4f})")

# Outliers IQR
q1, q3 = np.percentile(df["amount"], [25, 75])
iqr = q3 - q1
outliers = df[(df["amount"] < q1 - 1.5 * iqr) | (df["amount"] > q3 + 1.5 * iqr)]

Surveillance en continu : calculer les percentiles p50/p95/p99 par fenêtre temporelle et alerter si la variation dépasse ±20 % de la baseline glissante.

6. Data contracts

Structure minimale d'un contrat (YAML versionné dans le repo) :

contract_version: "1.2.0"
producer: "payment-service"
consumer: "analytics-pipeline"
schema_ref: "schemas/transactions_v3.avsc"
sla:
  freshness_minutes: 30
  completeness_pct: 99.5
  max_duplicates: 0
breaking_change_policy: "deprecation 30 jours, migration guide obligatoire"

Utiliser le Schema Registry (Confluent ou AWS Glue) pour enforcer côté broker :

# Vérifier la compatibilité avant deploy
curl -X POST http://schema-registry:8081/compatibility/subjects/transactions-value/versions/latest \
  -H "Content-Type: application/vnd.schemaregistry.v1+json" \
  -d '{"schema": "<nouveau_schema_json_escaped>"}'

7. Intégration pipeline — quarantaine et alertes

valid_rows   = df[validation_mask]
invalid_rows = df[~validation_mask].assign(quarantine_reason=error_series)

# Route vers dead-letter table
invalid_rows.to_sql("bad_data", engine, if_exists="append", index=False)

# Alerte si taux d'erreur dépasse le seuil
error_rate = len(invalid_rows) / len(df)
if error_rate > 0.02:   # seuil bloquant : 2 %
    raise PipelineError(f"Taux d'erreur critique : {error_rate:.1%}")
elif error_rate > 0.001: # warning : 0.1 %
    send_slack_alert(f"⚠️ Taux d'erreur : {error_rate:.1%} — vérifier bad_data")

8. Reporting et suivi

• Great Expectations Data Docs : générer avec context.build_data_docs(), exposer via S3 ou pages Git.
• Grafana : stocker les métriques de qualité dans Prometheus/InfluxDB, dashboards par pipeline.
• dbt artifacts : manifest.json + run_results.json → alimentation d'un catalogue (DataHub, Atlan).
• KPI à suivre : score de qualité global (%), volumétrie par statut (valid/quarantine/error), évolution sur 30 jours, top 5 violations par colonne.

Garde-fous et anti-patterns

Bonnes pratiques 2026

• Contract-first : définir le contrat avant d'écrire le pipeline, pas après.
• Shift-left : valider dès l'API d'ingestion (Pydantic), pas seulement dans le warehouse.
• Observabilité data : traiter la qualité comme un SLO — budget d'erreur, alertes PagerDuty si dépassement.
• Tests de non-régression : rejouer les jeux de données historiques après chaque changement de schéma.
• LLM-assisted validation (2025+) : utiliser un LLM pour générer des règles candidats depuis un échantillon de données, valider manuellement avant promotion en production.