Tutorial su Great Expectations: convalida dei dati con Python

Configurare Great Expectations

In questo tutorial imparerai a usare GX Core, la versione open source di Great Expectations, per convalidare un DataFrame Pandas. Vedremo come impostare un contesto, registrare una sorgente dati Pandas, definire le expectations e convalidare i batch di dati.

Nota: Ti consigliamo di seguire con il notebook DataLab, ma puoi anche creare il tuo script Python.

1. Installare Great Expectations

Prerequisiti

• Python dalla 3.9 alla 3.12 installato.
• Per evitare conflitti, è altamente consigliato installare Great Expectations in un ambiente virtuale (disclaimer: la configurazione degli ambienti virtuali è fuori dallo scopo di questo articolo).
• Un dataset di esempio.

Nota: Se usi il notebook DataLab fornito, questi prerequisiti sono già soddisfatti. Puoi saltarli.

Usa il seguente comando per installare GX tramite pip:

pip install great_expectations

Questo comando installa il pacchetto core e tutte le dipendenze necessarie.

2. Inizializzare il data context

Great Expectations richiede un data context per gestire le configurazioni. Usiamo un data context effimero per evitare di persistere le configurazioni.

import great_expectations as gx

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

Creare la tua prima suite di convalida dei dati

Ora che GX è configurato, creiamo una suite di convalida dei dati.

1. Connessione a una sorgente dati e creazione di un data asset

Una sorgente dati collega Great Expectations ai tuoi dati, mentre un data asset rappresenta un sottoinsieme specifico di dati (ad esempio, una tabella, un DataFrame o un file).

In questo caso, prepareremo tutto per connetterci a un DataFrame chiamato inventory_parts_df. Il dataset di esempio è disponibile nel DataLab fornito e viene creato una volta eseguito il blocco SQL:

Se non usi DataLab, crea un tuo DataFrame con dati di esempio.

Ora crea la tua sorgente dati e l'asset:

# 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. Aggiungere una definizione di batch

Una definizione di batch identifica e organizza i tuoi dati per la convalida. Qui aggiungiamo una definizione di batch che copre l’intero 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. Recuperare un batch

Un batch è una raccolta di dati collegata a una definizione di batch. Per convalidare i dati, dovrai recuperare e collegare il batch al tuo DataFrame, in questo caso 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. Creare una suite e definire le expectations

Le expectations sono regole per convalidare i dati. In questo esempio definiremo le seguenti semplici expectations:

1. Verificare che i valori di inventory_id non siano nulli.
2. Verificare che i valori di part_num siano unici.

# 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)

Puoi esplorare tutte le expectations disponibili nella Expectation Gallery. Ti incoraggiamo ad aggiungerne qualcuna in più!

Dopo aver definito le expectations, GX produce la configurazione della suite:

{
  "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 include i seguenti dettagli:

1. Nome e ID della suite: Un nome univoco (inventory_parts_suite) e un identificatore per tracciare e gestire la suite.
2. Expectations: Ogni regola specifica:
• Il tipo di controllo (ad esempio, assicurarsi che una colonna non abbia valori null o che contenga valori unici).
• I parametri, come la colonna da validare.
• Metadati e un ID univoco per ogni expectation, che ne facilitano tracciamento e personalizzazione.
3. Metadata: Informazioni sulla versione di Great Expectations, per garantire la compatibilità con lo strumento.
4. Note: Uno spazio per aggiungere commenti descrittivi sulla suite (opzionale).

Questo output strutturato funge sia da documentazione sia da configurazione riutilizzabile per convalidare il tuo dataset, così le expectations sono chiaramente definite, tracciabili e pronte per usi futuri.

5. Convalidare i dati

Infine, convalida il batch rispetto alle expectations definite e valuta i risultati.

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

Dopo l'esecuzione, Great Expectations fornisce un report dettagliato sul fatto che il dataset rispetti o meno le expectations definite:

{
  "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
}

Questo report dettaglia la qualità dei tuoi dati, evidenziando successi e fallimenti. Ecco una spiegazione semplificata dei risultati:

Convalida complessiva: Il risultato è stato parzialmente positivo: il 50% delle expectations è superato e il 50% è fallito. Un’expectation fallita indica un problema di qualità che richiede attenzione. In questo caso, una colonna non ha rispettato la regola definita.

Expectation 1: inventory_id non deve avere valori mancanti

• Risultato: Superato
• Spiegazione: Ogni valore nella colonna inventory_id è presente, senza voci null o mancanti. Questo indica una buona completezza dei dati per questa colonna.

Expectation 2: part_num deve avere valori unici

• Risultato: Fallito
• Spiegazione: La colonna part_num contiene il 97,98% di valori duplicati, il che significa che solo pochi valori sono unici.
• In evidenza:
• Esempi di valori duplicati includono "3069b" e "33291".
• Lo strumento mostra anche con quale frequenza compaiono questi duplicati e le loro posizioni di riga, facilitando l’individuazione e la correzione dei problemi.

Ovviamente questo è solo un dataset di esempio, e abbiamo inserito appositamente un’expectation superata e una fallita così da poter vedere entrambi i risultati di convalida.

Ecco fatto! Hai eseguito con successo convalide dei dati end-to-end.

Integrare Great Expectations nelle pipeline di dati

In produzione, le convalide devono essere integrate direttamente nel workflow per monitorare continuamente la qualità in ogni fase. 

In questa sezione vedremo come integrare Great Expectations nelle tue pipeline di dati.

Questi sono esempi per darti un’idea; potrebbero essere necessarie configurazioni aggiuntive non incluse qui. Controlla la documentazione di ciascuno strumento per la sintassi aggiornata!

Integrazione con strumenti ETL

Integrare Great Expectations con strumenti ETL popolari come Apache Airflow o Prefect è relativamente semplice. Inserire i passaggi di convalida direttamente nei processi ETL ti permette di individuare e risolvere i problemi in tempo reale, prima che incidano sulle analisi a valle.

Vediamo un semplice esempio di integrazione di Great Expectations con Prefect per eseguire la convalida come parte di un workflow ETL automatico:

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()

In questo esempio definiamo un flow Prefect con un task per eseguire la convalida con Great Expectations.

Il task validate_data() carica il contesto di Great Expectations, recupera il batch di dati e applica la suite di expectations. 

Se i dati non rispettano i criteri di convalida, il task genera un alert, interrompendo il workflow e prevenendo errori a valle.

Convalida continua dei dati

Puoi pianificare job di convalida usando vari strumenti, come cron sui sistemi Unix o servizi gestiti come Apache Airflow. In questo esempio mostreremo come pianificare esecuzioni di convalida con Airflow, ideale per orchestrare pipeline di dati.

Ecco come impostare un DAG (Directed Acyclic Graph) di Airflow per eseguire convalide Great Expectations ogni giorno:

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

In questo esempio definiamo un DAG che pianifica una convalida una volta al giorno (@daily). 

La funzione run_validation() esegue la convalida caricando il contesto di Great Expectations ed eseguendo la suite definita sui dati.

Best practice per la convalida dei dati con Great Expectations

Seguire le best practice è sempre consigliabile per scalabilità ed efficienza, e lo stesso vale per la convalida dei dati con Great Expectations.

Parti in piccolo e itera

Inizia con controlli fondamentali della qualità e amplia gradualmente. Meglio concentrarsi inizialmente su expectations di base per evitare di complicare troppo il processo, favorendo un’integrazione più fluida e un troubleshooting più semplice. Con l’aumentare della conoscenza del dataset, potrai aggiungere convalide più complesse.

Collabora tra i team

La qualità dei dati non è solo una questione tecnica. Collabora con i team di business per definire le expectations e assicurarti che le convalide implementate siano allineate alla logica e agli obiettivi aziendali. Questo approccio cross‑funzionale garantisce che i dati servano allo scopo previsto e soddisfino le esigenze di tutti gli stakeholder.

Automatizza quando possibile

Automatizza il processo ove possibile per integrare la convalida nelle pipeline di dati. L’integrazione di controlli automatici abilita un monitoraggio continuo della qualità senza intervento manuale, migliorando notevolmente l’efficienza.

Conclusione

Ottimo lavoro! Hai imparato a configurare e convalidare i dati in Great Expectations. Queste tecniche ti aiuteranno a mantenere alta qualità e trasparenza nei tuoi workflow.

Per continuare a sviluppare le tue competenze, dai un’occhiata a queste risorse:

• ETL and ELT in Python: impara a trasformare e spostare i dati in modo efficace.
• Introduction to Data Quality: scopri i fondamenti della gestione della qualità dei dati.
• Cleaning Data in Python: padroneggia le tecniche di pulizia per garantire accuratezza e coerenza.
• Data Quality Dimensions Cheat Sheet: una pratica guida alle dimensioni della qualità dei dati.