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 context, registrare una sorgente dati Pandas, definire expectations e convalidare batch di dati.

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

1. Installare Great Expectations

Prerequisiti

• Python 3.9 fino a 3.12 installato.
• Per evitare conflitti, è fortemente 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. Connettersi a una sorgente dati e creare un data asset

Una sorgente dati collega Great Expectations ai tuoi dati, mentre un data asset rappresenta un sottoinsieme specifico di dati (ad es., 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 il 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 batch definition

Una batch definition identifica e organizza i tuoi dati per la convalida. Qui aggiungiamo una batch definition 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 collezione di dati collegata a una batch definition. 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 expectations semplici:

1. Assicurati che i valori di inventory_id non siano nulli.
2. Assicurati che i valori di part_num siano univoci.

# 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 invitiamo ad aggiungerne qualcuna in più!

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

{
  "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 es., assicurarsi che una colonna non abbia valori nulli o che contenga voci uniche).
• I parametri, come la colonna oggetto della convalida.
• Metadati e un ID univoco per ciascuna expectation, per facilitarne tracciabilità e personalizzazione.
3. Metadata: informazioni sulla versione di Great Expectations, a garanzia della 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, in modo che le tue expectations siano 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 della convalida, Great Expectations fornisce un report dettagliato su quanto il dataset rispetti 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 illustra la qualità dei tuoi dati, evidenziando successi e fallimenti. Ecco una spiegazione semplificata dei risultati:

Convalida complessiva: Il risultato della convalida è stato parzialmente positivo: il 50% delle expectations è superato e il 50% fallito. Un'expectation fallita indica un problema di qualità dei dati 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 nulle o mancanti. Questo indica una buona completezza dei dati per questa colonna.

Expectation 2: part_num deve avere valori univoci

• 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 la frequenza di 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 volutamente un'expectation superata e una fallita per mostrarti entrambi i risultati della convalida.

È tutto! Hai eseguito con successo convalide dei dati end-to-end.

Integrare Great Expectations nelle pipeline di dati

In produzione, le convalide devono essere incorporate direttamente nel flusso di lavoro per monitorare continuamente la qualità dei dati 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. Consulta 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 intercettare e risolvere i problemi sui dati in tempo reale prima che impattino le analisi a valle.

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

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 di Prefect con un task per eseguire la convalida con Great Expectations.

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

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

Convalida continua dei dati

Puoi pianificare job di convalida usando vari strumenti, come cron su sistemi Unix o servizi gestiti come Apache Airflow. In questo esempio, mostreremo come pianificare esecuzioni di convalida con Airflow, particolarmente adatto all'orchestrazione di 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 un'esecuzione di convalida una volta al giorno (@daily). 

La funzione run_validation() esegue la convalida caricando il context di Great Expectations e applicando la suite di expectations definita ai dati.

Best practice per la convalida dei dati con Great Expectations

Seguire le best practice è sempre consigliato per scalabilità ed efficienza, e non è diverso per la convalida dei dati con Great Expectations.

Parti in piccolo e itera

Inizia con controlli fondamentali della qualità dei dati ed espandi gradualmente. È meglio concentrarsi inizialmente su expectations di base per evitare di complicare eccessivamente il processo, favorendo un'integrazione più fluida e una risoluzione dei problemi più semplice. Man mano che la tua comprensione del dataset migliora, puoi aggiungere convalide più complesse.

Collabora tra team

La qualità dei dati non è solo una preoccupazione tecnica. Collabora con i team di business per definire le expectations e assicurarti che la convalida implementata sia allineata alla logica e agli obiettivi aziendali. Questo approccio cross-funzionale garantisce che i dati servano allo scopo previsto e soddisfino i requisiti di tutti gli stakeholder.

Automatizza dove possibile

Automatizza il processo dove fattibile per integrare la convalida dei dati nelle pipeline. Integrare controlli di convalida automatici abilita il monitoraggio continuo della qualità dei dati senza interventi manuali, migliorando sensibilmente 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 dei dati nei tuoi workflow.

Per continuare a sviluppare le tue competenze, scopri queste risorse:

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