Guida alle Grandi Aspettative: Convalida dei Dati con Python

Configurazione di Great Expectations

In questo tutorial, imparerai come usare GX Core, la versione open-source di Great Expectations, per convalidare un DataFrame di Pandas. Ti guideremo attraverso la configurazione di un contesto, la registrazione di una sorgente dati Pandas, la definizione delle aspettative e la convalida dei batch di dati.

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

1. Installazione di Great Expectations

Requisiti

• Python 3.9 a 3.12 installato.
• Per evitare conflitti, è altamente consigliato installare Great Expectations all’interno di un ambiente virtuale (avviso: la configurazione degli ambienti virtuali è al di fuori dello scopo di questo articolo).
• Un set di dati di esempio.

Nota: Se si utilizza il notebook DataLab fornito, questi prerequisiti sono già stati soddisfatti. Sentiti libero di saltarli.

Usa il seguente comando per installare GX tramite pip:

pip install great_expectations

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

2. Inizializzazione del contesto dati

Great Expectations richiede un contesto dati per gestire le configurazioni. Utilizziamo un contesto dati effimero per evitare di persistere le configurazioni.

import great_expectations as gx

# Ottieni il Contesto Dati Effimero
context = gx.get_context()
assert type(context).__name__ == "EphemeralDataContext"

Creazione della tua prima suite di convalida dati

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

1. Connessione a una fonte di dati e creazione di un asset dati

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

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

Se non stai utilizzando DataLab, crea il tuo DataFrame con dati di esempio.

Ora, crea la tua fonte dati e risorsa:

# Aggiungi una Fonte Dati di Pandas
data_source = context.data_sources.add_pandas(name="inventory_parts")
# Aggiungi un Asset Dati alla Fonte Dati
data_asset = data_source.add_dataframe_asset(name="inventory_parts_asset")

2. Aggiunta di 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:

# Definire il nome della Definizione del Batch
batch_definition_name = "inventory_parts_batch"
# Aggiungere la Definizione del Batch
batch_definition = data_asset.add_batch_definition_whole_dataframe(batch_definition_name)
assert batch_definition.name == batch_definition_name

3. Recupero di un batch

Un batch è una raccolta di dati legata a una definizione del batch. Per convalidare i dati, è necessario recuperare e collegare il batch al DataFrame, in questo caso inventory_parts_df:

# Definire i Parametri del Batch
batch_parameters = {"dataframe": inventory_parts_df}
# Recuperare il Batch
batch = batch_definition.get_batch(batch_parameters=batch_parameters)

4. Creare un insieme e definire le aspettative

Le aspettative sono regole per la convalida dei dati. In questo esempio, definiremo le seguenti semplici aspettative:

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

# Crea una suite di aspettative
expectation_suite_name = "inventory_parts_suite"
suite = gx.ExpectationSuite(name=expectation_suite_name)
# Aggiungi aspettative
suite.add_expectation(
    gx.expectations.ExpectColumnValuesToNotBeNull(column="inventory_id")
)
suite.add_expectation(
    gx.expectations.ExpectColumnValuesToBeUnique(column="part_num")
)
# Aggiungi la suite di aspettative al contesto
context.suites.add(suite)

Puoi esplorare tutte le aspettative disponibili nella Galleria delle aspettative. Ti incoraggiamo ad aggiungerne qualcuna in più!

Dopo aver definito le aspettative, GX restituisce la configurazione della suite di aspettative:

{
  "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. Aspettative: Ogni regola specifica:
• Il tipo di controllo (ad esempio, assicurarsi che una colonna non abbia valori nulli o voci uniche).
• Parametri, come la colonna validata.
• Metadati e un ID univoco per ogni aspettativa, permettendo un tracciamento e una personalizzazione più facili.
3. Metadati: Informazioni sulla versione per Great Expectations, garantendo la compatibilità con lo strumento.
4. Note: Un segnaposto per aggiungere commenti descrittivi sul pacchetto (opzionale).

Questo output strutturato funge sia da documentazione che da configurazione riutilizzabile per convalidare il tuo set di dati in modo che le tue aspettative siano chiaramente definite, tracciabili e pronte per essere utilizzate in futuro.

5. Convalida dei dati

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

# Convalida i Dati Rispetto al Pacchetto
validation_results = batch.validate(suite)
# Valuta i Risultati
print(validation_results)

Dopo aver eseguito la convalida, Great Expectations fornisce un report dettagliato su se il set di dati soddisfa le aspettative 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 rapporto dettaglia la qualità dei tuoi dati, evidenziando successi e fallimenti. Ecco una spiegazione semplificata dei risultati:

Validazione complessiva: Il risultato della validazione è stato parzialmente positivo: il 50% delle aspettative è stato soddisfatto e il 50% è fallito. Un’aspettativa fallita indica un problema di qualità dei dati che richiede attenzione. In questo caso, una colonna non ha rispettato la regola definita.

Aspettativa 1: inventory_id non dovrebbe 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.

Aspettativa 2: part_num dovrebbe avere valori unici

• Risultato: Fallito
• Spiegazione: La colonna part_num contiene valori duplicati al 97,98%, il che significa che solo pochi valori sono unici.
• Evidenze:
• Esempi di valori duplicati includono “3069b” e “33291”.
• Lo strumento mostra anche con quale frequenza compaiono questi duplicati e le loro posizioni di riga, rendendo più facile individuare e risolvere i problemi.

Certo, questo è solo un set di dati di esempio e abbiamo incluso appositamente una aspettativa positiva e una negativa in modo che tu possa vedere entrambi i risultati della convalida.

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

Integrazione di Great Expectations nei Data Pipelines

In un ambiente di produzione, le convalidazioni devono essere incorporate direttamente nel flusso di lavoro per monitorare continuamente la qualità dei dati in ogni fase. 

In questa sezione, discuteremo su come puoi integrare Great Expectations nei tuoi data pipelines.

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. Incorporare i passaggi di convalida direttamente nei processi ETL ti permetterà di rilevare e affrontare i problemi di dati in tempo reale prima che influiscano sull’analisi successiva.

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

from prefect import task, Flow
import great_expectations as ge
# Definisci un'attività per eseguire la convalida di Great Expectations
@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])
	    
    # Controlla i risultati della convalida e invia un avviso se la convalida fallisce
    if not results["success"]:
        raise ValueError("Data validation failed!")
# Definisci il tuo flusso ETL
with Flow("ETL_with_GE_Validation") as flow:
    validation = validate_data()
# Esegui il flusso
flow.run()

In questo esempio, definiamo un flusso Prefect con un’attività per eseguire la convalida di Great Expectations.

Il compito validate_data() carica il contesto di Great Expectations, recupera il batch di dati e applica il suite di aspettative.

Se i dati non soddisfano i criteri di validazione, il compito solleva un avviso, interrompendo il flusso di lavoro e prevenendo errori a valle.

Validazione continua dei dati

Puoi pianificare lavori di validazione utilizzando vari strumenti, come i cron job su sistemi Unix-based o servizi gestiti come Apache Airflow. In questo esempio, dimostreremo come pianificare le esecuzioni di validazione utilizzando Airflow, che è ben adatto per orchestrare pipeline di dati.

Ecco come puoi impostare un DAG di Airflow (Grafo Direzionale Aciclico) per eseguire le validazioni di Great Expectations quotidianamente:

from airflow import DAG
from airflow.operators.python_operator import PythonOperator
from datetime import datetime
import great_expectations as ge
# Definire il DAG e impostare la pianificazione per l'esecuzione giornaliera
default_args = {
	  'owner': 'airflow',
	  'start_date': datetime(2024, 1, 1),
	  'retries': 1,
}
dag = DAG(
      'great_expectations_validation',
	default_args=default_args,
	schedule_interval='@daily',  # Eseguito una volta al giorno
)
# Definire la funzione per eseguire la validazione
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
# Configurare il compito in Airflow
validation_task = PythonOperator(
      task_id='run_great_expectations_validation',
      python_callable=run_validation,
      dag=dag,
)
# Impostare il compito nel DAG
validation_task

In questo esempio, definiamo un DAG che programma l’esecuzione di una validazione una volta al giorno (@daily).

La funzione run_validation() esegue la validazione caricando il contesto di Great Expectations ed eseguendo il set di aspettative definite sui dati.

Best Practices per la Validazione dei Dati con Great Expectations

Seguire le best practices è sempre consigliato per la scalabilità e l’efficienza, e non è diverso per la validazione dei dati con Great Expectations.

Cominciare in modo graduale ed iterare

Inizia con controlli di qualità dei dati fondamentali e espanditi gradualmente. È meglio concentrarsi inizialmente sulle aspettative di base, poiché ciò aiuta ad evitare di complicare eccessivamente il processo, rendendo l’integrazione più semplice e il troubleshooting più facile. Man mano che la comprensione del dataset migliora, è possibile aggiungere validazioni più complesse.

Collabora tra i team

La qualità dei dati non è solo una preoccupazione tecnica. Collabora tra i team aziendali per definire le aspettative e assicurarti che le validazioni implementate siano in linea con la logica aziendale sottostante e gli obiettivi. Questo approccio cross-funzionale garantisce che i dati servano lo scopo previsto e soddisfino i requisiti di tutti gli stakeholder.

Automatizza dove possibile

Automatizza il processo ovunque possibile per integrare la validazione dei dati nei data pipeline. Integrare controlli di validazione automatizzati consente il monitoraggio continuo della qualità dei dati senza intervento manuale, migliorando significativamente l’efficienza.

Conclusioni

Ottimo lavoro! Hai imparato come configurare e convalidare i dati in Great Expectations. Queste tecniche aiuteranno a mantenere alta la qualità e la trasparenza dei dati nei tuoi flussi di lavoro.

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

• ETL ed ELT in Python : Impara come trasformare e spostare i dati in modo efficace.
• Introduzione alla Qualità dei Dati : Esplora i fondamenti della gestione della qualità dei dati.
• Pulizia dei Dati in Python : Padrona le tecniche di pulizia dei dati per garantire precisione e coerenza.
• Cheat Sheet sulle Dimensioni della Qualità dei Dati : Una pratica guida alle dimensioni della qualità dei dati.