Great Expectations-tutorial: data valideren met Python

Great Expectations instellen

In deze tutorial leer je hoe je GX Core, de open-source versie van Great Expectations, gebruikt om een Pandas DataFrame te valideren. We lopen door het opzetten van een context, het registreren van een Pandas-databron, het definiëren van expectations en het valideren van databatches.

Let op: We raden aan om mee te doen met het DataLab-notebook, maar je kunt ook je eigen Python-script maken.

1. Great Expectations installeren

Vereisten

• Python 3.9 t/m 3.12 geïnstalleerd.
• Om conflicten te voorkomen raden we sterk aan Great Expectations in een virtuele omgeving te installeren (disclaimer: het opzetten van virtuele omgevingen valt buiten de scope van dit artikel).
• Een voorbeeld-dataset.

Let op: Als je het meegeleverde DataLab-notebook gebruikt, zijn deze vereisten al geregeld. Je kunt ze overslaan.

Gebruik het volgende commando om GX via pip te installeren:

pip install great_expectations

Dit commando installeert het kernpakket en alle benodigde afhankelijkheden.

2. De datacontext initialiseren

Great Expectations heeft een datacontext nodig om configuraties te beheren. We gebruiken een ephemeral datacontext om te voorkomen dat configuraties worden weggeschreven.

import great_expectations as gx

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

Je eerste datavalidatiesuite maken

Nu GX is ingesteld, maken we een datavalidatiesuite.

1. Verbinden met een databron en een data-asset maken

Een databron verbindt Great Expectations met je data, terwijl een data-asset een specifieke subset van data representeert (bijv. een tabel, DataFrame of bestand).

In dit geval bereiden we alles voor om te verbinden met een DataFrame genaamd inventory_parts_df. De voorbeeld-dataset is beschikbaar in de meegeleverde DataLab en wordt aangemaakt zodra we het SQL-blok uitvoeren:

Gebruik je DataLab niet, maak dan je eigen DataFrame met voorbeelddata.

Maak nu je databron en asset aan:

# 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. Een batchdefinitie toevoegen

Een batchdefinitie identificeert en organiseert je data voor validatie. Hier voegen we een batchdefinitie toe die het volledige DataFrame omvat:

# 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. Een batch ophalen

Een batch is een verzameling data gekoppeld aan een batchdefinitie. Om data te valideren moet je de batch ophalen en koppelen aan je DataFrame, in dit geval 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. Een suite maken en expectations definiëren

Expectations zijn regels voor het valideren van data. In dit voorbeeld definiëren we de volgende eenvoudige expectations:

1. Zorg dat inventory_id-waarden niet null zijn.
2. Zorg dat part_num-waarden uniek zijn.

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

Je kunt alle beschikbare expectations verkennen in de Expectation Gallery. Voeg er gerust nog een paar toe!

Na het definiëren van de expectations geeft GX de configuratie van de expectation suite weer:

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

De suite bevat de volgende details:

1. Suitenaam en ID: Een unieke naam (inventory_parts_suite) en identificator om de suite te volgen en beheren.
2. Expectations: Elke regel specificeert:
• Het type controle (bijv. zorgen dat een kolom geen null-waarden of unieke entries heeft).
• Parameters, zoals de te valideren kolom.
• Metadata en een unieke ID per expectation, wat het volgen en aanpassen vergemakkelijkt.
3. Metadata: Versie-informatie voor Great Expectations, om compatibiliteit met de tool te waarborgen.
4. Notities: Een placeholder om beschrijvende opmerkingen over de suite toe te voegen (optioneel).

Deze gestructureerde output fungeert zowel als documentatie als herbruikbare configuratie voor het valideren van je dataset, zodat je expectations duidelijk gedefinieerd, traceerbaar en klaar voor toekomstig gebruik zijn.

5. De data valideren

Valideer tot slot de batch aan de hand van de gedefinieerde expectations en evalueer de resultaten.

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

Na het uitvoeren van de validatie geeft Great Expectations een gedetailleerd rapport over de vraag of de dataset aan de gedefinieerde expectations voldoet:

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

Dit rapport beschrijft de kwaliteit van je data en markeert successen en mislukkingen. Hier is een vereenvoudigde uitleg van de resultaten:

Algemene validatie: Het validatieresultaat was deels geslaagd: 50% van de expectations is gehaald en 50% faalde. Een gefaalde expectation wijst op een datakwaliteitsprobleem dat aandacht nodig heeft. In dit geval voldeed één kolom niet aan de gedefinieerde regel.

Expectation 1: inventory_id mag geen missende waarden hebben

• Resultaat: Geslaagd
• Uitleg: Elke waarde in de kolom inventory_id is aanwezig, zonder null of missende entries. Dit duidt op goede datacompleetheid voor deze kolom.

Expectation 2: part_num moet unieke waarden hebben

• Resultaat: Gefaald
• Uitleg: De kolom part_num bevat 97,98% dubbele waarden, wat betekent dat slechts enkele waarden uniek zijn.
• Highlights:
• Voorbeelden van dubbele waarden zijn "3069b" en "33291".
• De tool laat ook zien hoe vaak deze duplicaten voorkomen en op welke rijen, zodat je ze makkelijker kunt vinden en oplossen.

Dit is natuurlijk slechts een voorbeeld-dataset; we hebben expres een geslaagde en een gefaalde expectation opgenomen zodat je beide validatieresultaten kunt zien.

Dat was het! Je hebt end-to-end datavalidaties succesvol uitgevoerd.

Great Expectations integreren in datapijplijnen

In een productie-omgeving moeten validaties direct in de workflow worden ingebed om de datakwaliteit continu in elke fase te monitoren. 

In deze sectie bespreken we hoe je Great Expectations integreert in je datapijplijnen.

Dit zijn voorbeelden om je een idee te geven; extra configuraties die hier niet zijn opgenomen kunnen nodig zijn. Raadpleeg de documentatie van elke tool voor de meest recente syntax!

Integratie met ETL-tools

Integreren van Great Expectations met populaire ETL-tools zoals Apache Airflow of Prefect is relatief eenvoudig. Door validatiestappen direct in de ETL-processen op te nemen, kun je dataproblemen in realtime opsporen en aanpakken voordat ze de analyses stroomafwaarts beïnvloeden.

Laten we een eenvoudig voorbeeld doornemen van integratie van Great Expectations met Prefect om datavalidatie uit te voeren als onderdeel van een geautomatiseerde ETL-werkstroom:

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 dit voorbeeld definiëren we een Prefect-flow met een taak voor het uitvoeren van Great Expectations-validatie.

De taak validate_data() laadt de Great Expectations-context, haalt de databatch op en past de expectation suite toe. 

Als de data niet aan de validatiecriteria voldoet, geeft de taak een waarschuwing en stopt de workflow, zodat fouten stroomafwaarts worden voorkomen.

Continue datavalidatie

Je kunt validatietaken inplannen met verschillende tools, zoals cronjobs op Unix-systemen of managed services zoals Apache Airflow. In dit voorbeeld laten we zien hoe je validaties plant met Airflow, dat zeer geschikt is voor het orkestreren van datapijplijnen.

Zo stel je een Airflow-DAG (Directed Acyclic Graph) in om dagelijks Great Expectations-validaties uit te voeren:

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 dit voorbeeld definiëren we een DAG die één keer per dag (@daily) een validatierun plant. 

De functie run_validation() voert de validatie uit door de Great Expectations-context te laden en de gedefinieerde expectation suite op de data toe te passen.

Best practices voor datavalidatie met Great Expectations

Het volgen van best practices is altijd aan te raden voor schaalbaarheid en efficiëntie, en dat is niet anders bij datavalidatie met Great Expectations.

Begin klein en iteratief

Begin met basale kwaliteitscontroles en breid geleidelijk uit. Richt je in eerste instantie op basis-expectations om overcomplicatie te voorkomen; dat zorgt voor soepelere integratie en eenvoudiger troubleshooting. Naarmate je de dataset beter leert kennen, kun je complexere validaties toevoegen.

Werk samen tussen teams

Datakwaliteit is niet alleen een technisch vraagstuk. Werk samen met business-teams om expectations te definiëren en te zorgen dat de implementatie aansluit bij de onderliggende businesslogica en doelen. Deze cross-functionele aanpak garandeert dat data haar beoogde doel dient en aan de eisen van alle stakeholders voldoet.

Automatiseer waar mogelijk

Automatiseer het proces waar dat kan om datavalidatie in je datapijplijnen te integreren. Geautomatiseerde validatiechecks maken continue monitoring van datakwaliteit mogelijk zonder handmatige tussenkomst, wat de efficiëntie aanzienlijk verbetert.

Conclusie

Goed gedaan! Je hebt geleerd hoe je data configureert en valideert in Great Expectations. Deze technieken helpen je hoge datakwaliteit en transparantie in je workflows te behouden.

Wil je je skills verder uitbouwen? Bekijk dan deze resources:

• ETL and ELT in Python: Leer hoe je data effectief transformeert en verplaatst.
• Introduction to Data Quality: Verken de basis van datakwaliteitsbeheer.
• Cleaning Data in Python: Beheers technieken voor datacleaning om nauwkeurigheid en consistentie te waarborgen.
• Data Quality Dimensions Cheat Sheet: Een handige gids voor datakwaliteitsdimensies.