Recette : Intégrer le contrôle de qualité des données d’entraînement avec Workflow v6

Contexte et objectifs

La qualité des données d’entraînement influence directement la robustesse et la performance des modèles d’intelligence artificielle. Dans un environnement d’ingénierie produit, il est essentiel de détecter les incohérences, les doublons ou les valeurs manquantes avant le lancement d’un job d’entraînement. Cette recette montre comment implémenter un contrôle de qualité (Data Quality Check, DQC) dans un pipeline Workflow v6, en s’appuyant sur des tâches Python, des déclencheurs d’alerte et la persistance des métriques.

1. Définir les règles de qualité

Avant d’écrire le code, listez les règles à appliquer :

• Complétude : chaque enregistrement doit contenir toutes les colonnes obligatoires.
• Valeurs hors‑plage : les champs numériques doivent rester dans les intervalles attendus.
• Duplication : aucune ligne ne doit être dupliquée.
• Consistance de type : les colonnes catégorielles ne doivent contenir que les catégories pré‑définies.

Ces règles sont décrites dans un fichier YAML (dqc_rules.yaml) afin de les rendre modifiables sans toucher au code :

# dqc_rules.yaml
completude:
  colonnes_obligatoires: ["id", "texte", "label"]
valeurs_hors_plage:
  age:
    min: 0
    max: 120
duplication:
  cle_primaire: "id"
consistance_type:
  label:
    categories: ["positif", "negatif", "neutre"]

2. Implémenter le contrôle dans le pipeline Workflow v6

2.1. Tâche Python de validation

Créez une tâche data_quality_check qui charge le jeu de données, applique les règles et renvoie un rapport JSON.

# tasks/data_quality_check.py
import yaml, json, pandas as pd
from pathlib import Path

def load_rules(path: str) -> dict:
    with open(path) as f:
        return yaml.safe_load(f)

def run_dqc(df: pd.DataFrame, rules: dict) -> dict:
    report = {"errors": []}
    # Complétude
    missing = [col for col in rules["completude"]["colonnes_obligatoires"] if col not in df.columns]
    if missing:
        report["errors"].append({"type": "missing_columns", "details": missing})
    # Valeurs hors‑plage
    for col, limits in rules.get("valeurs_hors_plage", {}).items():
        if col in df.columns:
            out_of_range = df[(df[col] < limits["min"]) | (df[col] > limits["max"])][col]
            if not out_of_range.empty:
                report["errors"].append({"type": "out_of_range", "column": col, "count": len(out_of_range)})
    # Duplication
    dup = df.duplicated(subset=rules["duplication"]["cle_primaire"], keep=False)
    if dup.any():
        report["errors"].append({"type": "duplicates", "count": dup.sum()})
    # Consistance de type
    for col, spec in rules.get("consistance_type", {}).items():
        if col in df.columns:
            invalid = ~df[col].isin(spec["categories"])
            if invalid.any():
                report["errors"].append({"type": "invalid_category", "column": col, "count": invalid.sum()})
    report["status"] = "OK" if not report["errors"] else "FAIL"
    return report

if __name__ == "__main__":
    df = pd.read_parquet(Path("{{input_path}}"))
    rules = load_rules("{{rules_path}}")
    report = run_dqc(df, rules)
    Path("{{output_path}}").write_text(json.dumps(report, indent=2))

2.2. Déclaration du workflow

Dans le fichier workflow.yaml, ajoutez la tâche et configurez les chemins d’entrée/sortie.

name: dqc-pipeline
steps:
  - name: fetch-data
    type: fetch
    args:
      uri: s3://bucket/dataset.parquet
      out: data.parquet

  - name: data-quality-check
    type: python
    args:
      script: tasks/data_quality_check.py
      input_path: data.parquet
      rules_path: dqc_rules.yaml
      output_path: dqc_report.json
    on_success: continue
    on_failure: abort

  - name: notify
    type: webhook
    args:
      url: https://hooks.example.com/alert
      payload: "{{ read_file('dqc_report.json') }}"
    condition: "{{ read_json('dqc_report.json').status == 'FAIL' }}"

3. Automatiser les alertes et les corrections

• Alertes : le step notify ne s’exécute que si le rapport indique un statut FAIL. Le payload JSON contient la liste des erreurs, facilitant le triage.
• Correction automatique : pour les cas simples (ex. valeurs manquantes), on peut ajouter un step impute-missing qui remplit les NaN avec la moyenne ou une valeur par défaut. Ce step doit être placé entre fetch-data et data-quality-check et conditionné par la présence d’erreurs de type missing_columns.
• Persistencia : stockez les rapports DQC dans un bucket dédié (s3://bucket/dqc-reports/) avec un horodatage. Cela crée un historique exploitable pour l’audit et le suivi de la dérive de données.

4. Bonnes pratiques et pièges à éviter

• Versionner les règles : conservez le fichier dqc_rules.yaml dans le même dépôt que le workflow. Chaque modification doit être revue comme tout changement de code.
• Idempotence : assurez‑vous que la tâche data_quality_check ne modifie pas le jeu de données. Elle ne doit produire que des métriques, pour éviter les effets de bord.
• Scalabilité : pour des jeux de données supérieurs à quelques gigaoctets, utilisez le mode pandas.read_parquet(..., engine='pyarrow') avec le paramètre chunksize ou migrez vers Spark via le composant spark-submit de Workflow v6.
• Surveillance des temps d’exécution : un DQC très coûteux peut retarder l’ensemble du pipeline. Mesurez le temps de chaque step et définissez un seuil d’alerte (max_duration: 300 s) dans le fichier workflow.yaml.
• Gestion des faux positifs : certaines règles peuvent générer des alertes non critiques. Implémentez un mécanisme de sévérité (warning vs error) et filtrez les notifications en conséquence.

En intégrant ces contrôles de qualité directement dans le pipeline Workflow v6, les équipes produit gagnent en confiance sur leurs données, réduisent les cycles de ré‑entraînement coûteux et améliorent la stabilité des modèles en production.