Didacticiel sur les grandes espérances : Valider des données avec Python

Des attentes élevées

Dans ce tutoriel, vous apprendrez à utiliser GX Core, la version open-source de Great Expectations, pour valider un DataFrame Pandas. Nous verrons comment configurer un contexte, enregistrer une source de données Pandas, définir les attentes et valider les lots de données.

Note : Nous vous recommandons de suivre le carnet DataLab, mais vous pouvez également créer votre propre script Python.

1. Installer les grandes espérances

Conditions préalables

• Python 3.9 à 3.12 installé.
• Pour éviter les conflits, il est fortement recommandé d'installer Great Expectations dans un environnement virtuel (avis de non-responsabilité : l'installation d'environnements virtuels dépasse le cadre de cet article).
• Un échantillon de données.

Note : Si vous utilisez le carnet DataLab fourni, ces conditions préalables ont déjà été remplies. N'hésitez pas à les ignorer.

Utilisez la commande suivante pour installer GX via pip :

pip install great_expectations

Cette commande installe le paquetage de base et toutes les dépendances nécessaires.

2. Initialisation du contexte de données

Great Expectations nécessite un contexte de données pour gérer les configurations. Nous utilisons un contexte de données éphémère pour éviter la persistance des configurations.

import great_expectations as gx

# Get the Ephemeral Data Context
context = gx.get_context()
assert type(context).__name__ == "EphemeralDataContext"

Créer votre première suite de validation des données

Maintenant que GX est configuré, créons une suite de validation des données.

1. Connexion à une source de données et création d'une ressource de données

Une source de données relie Great Expectations à vos données, tandis qu'une ressource de données représente un sous-ensemble spécifique de données (par exemple, un tableau, un DataFrame ou un fichier).

Dans ce cas, nous allons tout préparer pour nous connecter à un DataFrame appelé inventory_parts_df. L'échantillon de données est disponible dans le DataLab fourni, et il est créé une fois que nous exécutons le bloc SQL :

Si vous n'utilisez pas DataLab, créez votre propre DataFrame avec des échantillons de données.

Créez maintenant votre source de données et votre ressource :

# Add a Pandas Data Source
data_source = context.data_sources.add_pandas(name="inventory_parts")
# Add a Data Asset to the Data Source
data_asset = data_source.add_dataframe_asset(name="inventory_parts_asset")

2. Ajout d'une définition de lot

Une définition de lot identifie et organise vos données pour la validation. Ici, nous ajoutons une définition de lot qui couvre l'ensemble du DataFrame :

# Define the Batch Definition name
batch_definition_name = "inventory_parts_batch"
# Add the Batch Definition
batch_definition = data_asset.add_batch_definition_whole_dataframe(batch_definition_name)
assert batch_definition.name == batch_definition_name

3. Récupération d'un lot

Un lot est un ensemble de données liées à une définition de lot. Pour valider les données, vous devez récupérer le lot et le lier à votre DataFrame, dans ce cas inventory_parts_df:

# Define the Batch Parameters
batch_parameters = {"dataframe": inventory_parts_df}
# Retrieve the Batch
batch = batch_definition.get_batch(batch_parameters=batch_parameters)

4. Créer une suite et définir les attentes

Les attentes sont des règles de validation des données. Dans cet exemple, nous définirons les attentes simples suivantes :

1. Assurez-vous que les valeurs de inventory_id sont non nulles.
2. Assurez-vous que les valeurs de part_num sont uniques.

# Create an Expectation Suite
expectation_suite_name = "inventory_parts_suite"
suite = gx.ExpectationSuite(name=expectation_suite_name)
# Add Expectations
suite.add_expectation(
    gx.expectations.ExpectColumnValuesToNotBeNull(column="inventory_id")
)
suite.add_expectation(
    gx.expectations.ExpectColumnValuesToBeUnique(column="part_num")
)
# Add the Expectation Suite to the Context
context.suites.add(suite)

Vous pouvez découvrir toutes les attentes disponibles dans la galerie des attentes. Nous vous encourageons à en ajouter d'autres !

Après avoir défini les attentes, le GX produit la configuration de la suite d'attentes :

{
  "name": "inventory_parts_suite",
  "id": "b2de0b69-0869-4163-8dde-6c09884483f7",
  "expectations": [
    {
      "type": "expect_column_values_to_not_be_null",
      "kwargs": {
        "column": "inventory_id"
      },
      "meta": {},
      "id": "53d6c42a-d190-412f-a113-783b706531f4"
    },
    {
      "type": "expect_column_values_to_be_unique",
      "kwargs": {
        "column": "part_num"
      },
      "meta": {},
      "id": "362a2bdc-616d-4b3a-b7f0-c73808caee78"
    }
  ],
  "meta": {
    "great_expectations_version": "1.2.4"
  },
  "notes": null
}

La suite comprend les éléments suivants :

1. Nom et numéro d'identification de la suite : Un nom unique (inventory_parts_suite) et un identifiant pour suivre et gérer la suite.
2. Attentes : Chaque règle spécifie
• Le type de contrôle (par exemple, s'assurer qu'une colonne ne contient pas de valeurs nulles ou d'entrées uniques).
• Paramètres, tels que la colonne à valider.
• Des cursus et un identifiant unique pour chaque attente, permettant un suivi et une personnalisation plus aisés.
3. Métadonnées : Informations sur la version de Great Expectations, garantissant la compatibilité avec l'outil.
4. Notes : Un espace réservé pour ajouter des commentaires descriptifs sur la suite (facultatif).

Ce résultat structuré sert à la fois de documentation et de configuration réutilisable pour valider votre ensemble de données afin que vos attentes soient clairement définies, traçables et prêtes à être utilisées à l'avenir.

5. Validation des données

Enfin, validez le lot par rapport aux attentes définies et évaluez les résultats.

# Validate the Data Against the Suite
validation_results = batch.validate(suite)
# Evaluate the Results
print(validation_results)

Après avoir effectué la validation, Great Expectations fournit un rapport détaillé indiquant si l'ensemble de données répond aux attentes définies :

{
  "success": false,
  "results": [
    {
      "success": true,
      "expectation_config": {
        "type": "expect_column_values_to_not_be_null",
        "kwargs": {
          "batch_id": "inventory_parts-inventory_parts_asset",
          "column": "inventory_id"
        },
        "meta": {},
        "id": "53d6c42a-d190-412f-a113-783b706531f4"
      },
      "result": {
        "element_count": 580069,
        "unexpected_count": 0,
        "unexpected_percent": 0.0,
        "partial_unexpected_list": [],
        "partial_unexpected_counts": [],
        "partial_unexpected_index_list": []
      },
      "meta": {},
      "exception_info": {
        "raised_exception": false,
        "exception_traceback": null,
        "exception_message": null
      }
    },
    {
      "success": false,
      "expectation_config": {
        "type": "expect_column_values_to_be_unique",
        "kwargs": {
          "batch_id": "inventory_parts-inventory_parts_asset",
          "column": "part_num"
        },
        "meta": {},
        "id": "362a2bdc-616d-4b3a-b7f0-c73808caee78"
      },
      "result": {
        "element_count": 580069,
        "unexpected_count": 568352,
        "unexpected_percent": 97.98006788847535,
        "partial_unexpected_list": [
          "48379c01",
          "paddle",
          "11816pr0005",
          "2343",
          "3003",
          "30176",
          "3020",
          "3022",
          "3023",
          "30357",
          "3039",
          "3062b",
          "3068b",
          "3069b",
          "3069b",
          "33291",
          "33291",
          "3795",
          "3941",
          "3960"
        ],
        "missing_count": 0,
        "missing_percent": 0.0,
        "unexpected_percent_total": 97.98006788847535,
        "unexpected_percent_nonmissing": 97.98006788847535,
        "partial_unexpected_counts": [
          {
            "value": "3069b",
            "count": 2
          },
          {
            "value": "33291",
            "count": 2
          },
          {
            "value": "11816pr0005",
            "count": 1
          },
          {
            "value": "2343",
            "count": 1
          },
          {
            "value": "3003",
            "count": 1
          },
          {
            "value": "30176",
            "count": 1
          },
          {
            "value": "3020",
            "count": 1
          },
          {
            "value": "3022",
            "count": 1
          },
          {
            "value": "3023",
            "count": 1
          },
          {
            "value": "30357",
            "count": 1
          },
          {
            "value": "3039",
            "count": 1
          },
          {
            "value": "3062b",
            "count": 1
          },
          {
            "value": "3068b",
            "count": 1
          },
          {
            "value": "3795",
            "count": 1
          },
          {
            "value": "3941",
            "count": 1
          },
          {
            "value": "3960",
            "count": 1
          },
          {
            "value": "48379c01",
            "count": 1
          },
          {
            "value": "paddle",
            "count": 1
          }
        ],
        "partial_unexpected_index_list": [
          0,
          3,
          4,
          5,
          6,
          7,
          8,
          9,
          10,
          11,
          12,
          13,
          14,
          15,
          16,
          17,
          18,
          19,
          20,
          21
        ]
      },
      "meta": {},
      "exception_info": {
        "raised_exception": false,
        "exception_traceback": null,
        "exception_message": null
      }
    }
  ],
  "suite_name": "inventory_parts_suite",
  "suite_parameters": {},
  "statistics": {
    "evaluated_expectations": 2,
    "successful_expectations": 1,
    "unsuccessful_expectations": 1,
    "success_percent": 50.0
  },
  "meta": {
    "great_expectations_version": "1.2.4",
    "batch_spec": {
      "batch_data": "PandasDataFrame"
    },
    "batch_markers": {
      "ge_load_time": "20241129T122532.416424Z",
      "pandas_data_fingerprint": "84a1e1939091fcf54324910def3b89cd"
    },
    "active_batch_definition": {
      "datasource_name": "inventory_parts",
      "data_connector_name": "fluent",
      "data_asset_name": "inventory_parts_asset",
      "batch_identifiers": {
        "dataframe": "<DATAFRAME>"
      }
    }
  },
  "id": null
}

Ce rapport détaille la qualité de vos données, en soulignant les succès et les échecs. Voici une explication simplifiée des résultats :

Validation globale : Le résultat de la validation a été partiellement satisfaisant : 50 % des attentes ont été satisfaites et 50 % ont échoué. Une attente non satisfaite indique un problème de qualité des données qui nécessite une attention particulière. Dans ce cas, une colonne ne répondait pas à la règle définie.

Attente 1 : inventory_id ne pas avoir de valeurs manquantes

• Résultat: Adopté
• Explication: Toutes les valeurs de la colonne inventory_id sont présentes, sans aucune entrée nulle ou manquante. Cela indique que les données sont bien complètes pour cette colonne.

Attente 2 : part_num doivent avoir des valeurs uniques

• Résultat : Échec
• Explication : La colonne part_num contient 97,98 % de valeurs dupliquées, ce qui signifie que seules quelques valeurs sont uniques.
• Faits marquants :
• Les valeurs dupliquées sont par exemple "3069b" et "33291".
• L'outil indique également la fréquence d'apparition de ces doublons et leur position sur la ligne, ce qui facilite la localisation et la résolution des problèmes.

Bien entendu, il ne s'agit que d'un échantillon de données, et nous avons volontairement inclus une attente de réussite et une attente d'échec afin que vous puissiez voir les deux résultats de validation.

C'est tout ! Vous avez effectué avec succès des validations de données de bout en bout.

Intégrer les grandes attentes dans les pipelines de données

Dans un contexte de production, les validations doivent être intégrées directement dans le flux de travail afin de contrôler en permanence la qualité des données à chaque étape. 

Dans cette section, nous verrons comment vous pouvez intégrer Great Expectations dans vos pipelines de données.

Il s'agit d'exemples pour vous donner une idée, et des configurations supplémentaires non incluses ici peuvent être nécessaires. Consultez la documentation de chaque outil pour connaître la syntaxe la plus récente !

Intégration avec les outils ETL

L'intégration de Great Expectations avec des outils ETL populaires comme Apache Airflow ou Prefect est relativement simple. L'intégration d'étapes de validation directement dans les processus ETL vous permettra de détecter et de traiter les problèmes de données en temps réel avant qu'ils n'affectent l'analyse en aval.

Voyons un exemple simple d'intégration de Great Expectations avec Prefect pour exécuter la validation des données dans le cadre d'un flux de travail ETL automatisé :

from prefect import task, Flow
import great_expectations as ge
# Define a task to run Great Expectations validation
@task
def validate_data():
    context = ge.data_context.DataContext()
    batch_kwargs = {"path": "path/to/your/datafile.csv", "datasource": "your_datasource"}
    batch = context.get_batch(batch_kwargs, suite_name="your_expectation_suite")
    results = context.run_validation_operator("action_list_operator", assets_to_validate=[batch])
	    
    # Check validation results and raise an alert if validation fails
    if not results["success"]:
        raise ValueError("Data validation failed!")
# Define your ETL flow
with Flow("ETL_with_GE_Validation") as flow:
    validation = validate_data()
# Execute the flow
flow.run()

Dans cet exemple, nous définissons un flux Préfet avec une tâche de validation des Grandes Espérances.

La tâche validate_data() charge le contexte Great Expectations, récupère le lot de données et applique la suite d'attentes. 

Si les données ne répondent pas aux critères de validation, la tâche déclenche une alerte, ce qui interrompt le flux de travail et évite les erreurs en aval.

Validation continue des données

Vous pouvez planifier les tâches de validation à l'aide de différents outils, tels que les tâches cron sur les systèmes Unix ou les services gérés comme Apache Airflow. Dans cet exemple, nous allons montrer comment planifier des cycles de validation à l'aide d'Airflow, qui est bien adapté à l'orchestration des pipelines de données.

Voici comment vous pouvez mettre en place un DAG(Directed Acyclic Graph) Airflow pour exécuter quotidiennement les validations Great Expectations :

from airflow import DAG
from airflow.operators.python_operator import PythonOperator
from datetime import datetime
import great_expectations as ge
# Define the DAG and set the schedule to run daily
default_args = {
	  'owner': 'airflow',
	  'start_date': datetime(2024, 1, 1),
	  'retries': 1,
}
dag = DAG(
      'great_expectations_validation',
	default_args=default_args,
	schedule_interval='@daily',  # Runs once a day
)
# Define the function to run the validation
def run_validation():
    context = ge.data_context.DataContext()
    batch = context.get_batch(batch_kwargs, suite_name="your_expectation_suite")
    results = context.run_validation_operator("action_list_operator", assets_to_validate=[batch])
    return results
# Set up the task in Airflow
validation_task = PythonOperator(
      task_id='run_great_expectations_validation',
      python_callable=run_validation,
      dag=dag,
)
# Set the task in the DAG
validation_task

Dans cet exemple, nous définissons un DAG qui programme un cycle de validation une fois par jour (@daily). 

La fonction run_validation() exécute la validation en chargeant le contexte Great Expectations et en exécutant la suite d'attentes définie par rapport aux données.

Meilleures pratiques pour la validation des données avec de grandes attentes

Il est toujours préférable de suivre les meilleures pratiques pour assurer l'évolutivité et l'efficacité, et il en va de même pour la validation des données avec Great Expectations.

Commencez petit et répétez

Commencez par des contrôles fondamentaux de la qualité des données et étendez-les progressivement. Il est préférable de se concentrer sur les attentes fondamentales au départ, car cela permet d'éviter de trop compliquer le processus, ce qui facilite l'intégration et le dépannage. Au fur et à mesure que votre compréhension de l'ensemble de données s'améliore, vous pouvez ajouter des validations plus complexes.

Collaborer entre équipes

La qualité des données n'est pas seulement une question technique. Collaborer avec les équipes commerciales pour définir les attentes et veiller à ce que la validation mise en œuvre corresponde à la logique et aux objectifs commerciaux sous-jacents. Cette approche interfonctionnelle garantit que les données servent l'objectif visé et répondent aux exigences de toutes les parties prenantes.

Automatiser dans la mesure du possible

Automatisez le processus dans la mesure du possible afin d'intégrer la validation des données dans les pipelines de données. L'intégration de contrôles de validation automatisés permet de surveiller en permanence la qualité des données sans intervention manuelle, ce qui améliore considérablement l'efficacité.

Conclusion

Excellent travail ! Vous avez appris à configurer et à valider les données dans Great Expectations. Ces techniques vous aideront à maintenir la qualité des données et la transparence dans vos flux de travail.

Pour continuer à développer vos compétences, consultez ces ressources :

• ETL et ELT en Python: Apprenez à transformer et à déplacer les données de manière efficace.
• Introduction à la qualité des données: Découvrez les principes fondamentaux de la gestion de la qualité des données.
• Nettoyage de données en Python: Techniques de nettoyage des données de base pour garantir l'exactitude et la cohérence.
• Aide-mémoire sur les dimensions de la qualité des données: Un guide pratique sur les dimensions de la qualité des données.