Tutorial Great Expectations: Validarea datelor cu Python

Configurarea Great Expectations

În acest tutorial, vei învăța să folosești GX Core, versiunea open-source a Great Expectations, pentru a valida un Pandas DataFrame. Vom parcurge configurarea unui context, înregistrarea unei surse de date Pandas, definirea „expectations” și validarea loturilor de date.

Notă: Îți recomandăm să urmărești împreună cu caietul DataLab, dar poți crea și propriul script Python.

1. Instalarea Great Expectations

Cerințe preliminare

• Python 3.9–3.12 instalat.
• Pentru a evita conflictele, este recomandat cu tărie să instalezi Great Expectations într-un mediu virtual (disclaimer: configurarea mediilor virtuale depășește scopul acestui articol).
• Un set de date exemplu.

Notă: Dacă folosești caietul DataLab furnizat, aceste cerințe sunt deja îndeplinite. Poți să le sari.

Folosește următoarea comandă pentru a instala GX prin pip:

pip install great_expectations

Această comandă instalează pachetul de bază și toate dependențele necesare.

2. Inițializarea contextului de date

Great Expectations are nevoie de un context de date pentru a gestiona configurațiile. Folosim un context efemer pentru a evita persistarea configurațiilor.

import great_expectations as gx

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

Crearea primei tale suite de validare a datelor

Acum că GX este configurat, să creăm o suită de validare a datelor.

1. Conectarea la o sursă de date și crearea unui asset

O sursă de date conectează Great Expectations la datele tale, în timp ce un asset de date reprezintă un subset specific de date (de ex., un tabel, un DataFrame sau un fișier).

În acest caz, vom pregăti totul pentru a ne conecta la un DataFrame numit inventory_parts_df. Setul de date exemplu este disponibil în DataLab-ul furnizat și este creat odată ce rulăm blocul SQL:

Dacă nu folosești DataLab, creează-ți propriul DataFrame cu date exemplu.

Acum, creează sursa și assetul de date:

# 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. Adăugarea unei definiții de batch

O definiție de batch identifică și organizează datele pentru validare. Aici, adăugăm o definiție care acoperă întregul 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. Preluarea unui batch

Un batch este o colecție de date legată de o definiție de batch. Pentru a valida datele, va trebui să preiei și să legi batch-ul de DataFrame-ul tău, în acest caz 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. Crearea unei suite și definirea „expectations”

„Expectations” sunt reguli pentru validarea datelor. În acest exemplu, vom defini următoarele reguli simple:

1. Asigură-te că valorile inventory_id nu sunt nule.
2. Asigură-te că valorile part_num sunt unice.

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

Poți explora toate „expectations” disponibile în Expectation Gallery. Te încurajăm să mai adaugi câteva!

După definirea regulilor, GX generează configurația suitei de „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
}

Suita include următoarele detalii:

1. Numele și ID-ul suitei: Un nume unic (inventory_parts_suite) și un identificator pentru gestionarea și urmărirea suitei.
2. „Expectations”: Fiecare regulă specifică:
• Tipul verificării (de ex., asigurarea că o coloană nu are valori nule sau că are intrări unice).
• Parametri, precum coloana validată.
• Metadate și un ID unic pentru fiecare regulă, pentru urmărire și personalizare mai ușoară.
3. Metadate: Informații despre versiunea Great Expectations, pentru a asigura compatibilitatea.
4. Note: Un loc pentru comentarii descriptive despre suită (opțional).

Acest output structurat servește atât ca documentație, cât și ca o configurație reutilizabilă pentru validarea setului tău de date, astfel încât regulile să fie clar definite, trasabile și gata de folosit în viitor.

5. Validarea datelor

În cele din urmă, validează batch-ul în raport cu regulile definite și evaluează rezultatele.

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

După rularea validării, Great Expectations furnizează un raport detaliat privind măsura în care setul de date îndeplinește regulile 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
}

Acest raport detaliază calitatea datelor tale, evidențiind reușitele și eșecurile. Iată o explicație simplificată a rezultatelor:

Validare generală: Rezultatul a fost parțial reușit: 50% dintre reguli au trecut și 50% au eșuat. O regulă eșuată indică o problemă de calitate a datelor care necesită atenție. În acest caz, o coloană nu a respectat regula definită.

Expectation 1: inventory_id nu trebuie să aibă valori lipsă

• Rezultat: A trecut
• Explicație: Fiecare valoare din coloana inventory_id este prezentă, fără intrări nule sau lipsă. Acest lucru indică o bună completitudine a datelor pentru această coloană.

Expectation 2: part_num trebuie să aibă valori unice

• Rezultat: A eșuat
• Explicație: Coloana part_num conține 97,98% valori duplicate, ceea ce înseamnă că doar câteva valori sunt unice.
• Evidențieri:
• Exemple de valori duplicate includ „3069b” și „33291”.
• Instrumentul arată și frecvența acestor duplicate și pozițiile lor pe rânduri, facilitând localizarea și corectarea problemelor.

Desigur, acesta este doar un set de date de exemplu, și am inclus intenționat o regulă care trece și una care eșuează, ca să poți vedea ambele rezultate.

Gata! Ai rulat cu succes validări de date cap-coadă.

Integrarea Great Expectations în pipeline-uri de date

În producție, validările trebuie integrate direct în fluxul de lucru pentru a monitoriza continuu calitatea datelor în fiecare etapă.

În această secțiune, vom discuta cum poți integra Great Expectations în pipeline-urile tale de date.

Acestea sunt exemple pentru orientare, iar configurații suplimentare care nu sunt incluse aici pot fi necesare. Consultă documentația fiecărui instrument pentru sintaxa actualizată!

Integrare cu instrumente ETL

Integrarea Great Expectations cu instrumente ETL populare precum Apache Airflow sau Prefect este relativ simplă. Includerea pașilor de validare direct în procesele ETL îți va permite să identifici și să remediezi problemele de date în timp real, înainte să afecteze analizele ulterioare.

Să parcurgem un exemplu simplu de integrare a Great Expectations cu Prefect pentru a rula validarea datelor ca parte a unui workflow ETL automatizat:

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

În acest exemplu, definim un flow Prefect cu o sarcină pentru rularea validării Great Expectations.

Funcția validate_data() încarcă contextul Great Expectations, preia batch-ul de date și aplică suita de reguli.

Dacă datele nu îndeplinesc criteriile de validare, sarcina ridică o alertă, oprind workflow-ul și prevenind erorile din etapele următoare.

Validare continuă a datelor

Poți programa joburi de validare folosind diverse instrumente, precum cron pe sisteme Unix sau servicii gestionate ca Apache Airflow. Pentru acest exemplu, vom arăta cum să programezi rulări de validare cu Airflow, potrivit pentru orchestrarea pipeline-urilor de date.

Iată cum poți configura un DAG Airflow (Directed Acyclic Graph) pentru a rula validări Great Expectations zilnic:

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

În acest exemplu, definim un DAG care programează o rulare de validare o dată pe zi (@daily).

Funcția run_validation() execută validarea prin încărcarea contextului Great Expectations și rularea suitei de reguli definite asupra datelor.

Bune practici pentru validarea datelor cu Great Expectations

Respectarea bunelor practici este întotdeauna recomandată pentru scalabilitate și eficiență, iar validarea datelor cu Great Expectations nu face excepție.

Începe mic și iterează

Începe cu verificări fundamentale ale calității datelor și extinde treptat. E mai bine să te concentrezi inițial pe reguli de bază, pentru a evita complicarea procesului — astfel integrarea va fi mai lină și depanarea mai ușoară. Pe măsură ce îți înțelegi mai bine setul de date, poți adăuga validări mai complexe.

Colaborează între echipe

Calitatea datelor nu este doar o preocupare tehnică. Colaborează cu echipele de business pentru a defini regulile și pentru a te asigura că validarea implementată se aliniază cu logica și obiectivele de business. Această abordare cross-funcțională garantează că datele își servesc scopul și îndeplinesc cerințele tuturor părților interesate.

Automatizează acolo unde este posibil

Automatizează procesul ori de câte ori este fezabil, pentru a integra validarea în pipeline-urile de date. Validările automate permit monitorizarea continuă a calității fără intervenție manuală, îmbunătățind semnificativ eficiența.

Concluzie

Foarte bine! Ai învățat cum să configurezi și să validezi date în Great Expectations. Aceste tehnici te vor ajuta să menții o calitate ridicată a datelor și transparență în fluxurile tale de lucru.

Pentru a-ți continua dezvoltarea, consultă aceste resurse:

• ETL și ELT în Python: Învață să transformi și să muți datele eficient.
• Introducere în calitatea datelor: Explorează elementele de bază ale managementului calității datelor.
• Curățarea datelor în Python: Stăpânește tehnici de curățare pentru acuratețe și consistență.
• Fișă de sinteză: Dimensiuni ale calității datelor: Un ghid util pentru dimensiunile calității datelor.