Great Expectations-tutorial: data valideren met Python

Great Expectations instellen

In deze tutorial leer je hoe je GX Core, de open-sourceversie van Great Expectations, gebruikt om een Pandas DataFrame te valideren. We lopen door het instellen 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 tot en met 3.12 geïnstalleerd.
• Om conflicten te vermijden, raden we sterk aan om 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 core-pakket en alle benodigde dependencies.

2. De datacontext initialiseren

Great Expectations heeft een datacontext nodig om configuraties te beheren. We gebruiken een vluchtige (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 vertegenwoordigt (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 die is 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 vooral 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 identifier om de suite te beheren en te volgen.
2. Expectations: Elke regel specificeert:
• Het type controle (bijv. garanderen dat een kolom geen null-waarden of alleen unieke waarden bevat).
• Parameters, zoals de te valideren kolom.
• Metadata en een uniek ID per expectation, wat het volgen en aanpassen vergemakkelijkt.
3. Metadata: Versie-informatie voor Great Expectations, voor compatibiliteit met de tool.
4. Notities: Een plek voor beschrijvende opmerkingen over de suite (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 ten slotte de batch tegen 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 met 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 licht successen en fouten uit. Hier is een vereenvoudigde uitleg van de resultaten:

Algehele validatie: De validatie was deels geslaagd: 50% van de expectations slaagde en 50% faalde. Een mislukte expectation wijst op een datakwaliteitsprobleem dat aandacht vereist. In dit geval voldeed één kolom niet aan de gedefinieerde regel.

Expectation 1: inventory_id mag geen missende waarden bevatten

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

Expectation 2: part_num moet unieke waarden hebben

• Resultaat: Mislukt
• Uitleg: De kolom part_num bevat 97,98% duplicate waarden, wat betekent dat slechts enkele waarden uniek zijn.
• Highlights:
• Voorbeelden van duplicate waarden zijn "3069b" en "33291".
• De tool toont ook hoe vaak deze duplicaten voorkomen en hun rijposities, waardoor je de problemen makkelijker kunt lokaliseren en oplossen.

Dit is natuurlijk slechts een voorbeeld-dataset, en we hebben opzettelijk één geslaagde en één mislukte expectation opgenomen zodat je beide validatieresultaten ziet.

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

Great Expectations integreren in datapijplijnen

In een productieomgeving moeten validaties direct in de workflow ingebed zijn om de datakwaliteit continu in elke fase te monitoren. 

In deze sectie bespreken we hoe je Great Expectations kunt integreren 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 je ETL-processen op te nemen, kun je dataproblemen in realtime opvangen en oplossen voordat ze de analyses downstream beïnvloeden.

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

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 om Great Expectations-validatie uit te voeren.

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 alert en stopt de workflow, zodat fouten downstream worden voorkomen.

Continue datavalidatie

Je kunt validatiejobs plannen met diverse tools, zoals cronjobs op Unix-systemen of beheerde services zoals Apache Airflow. In dit voorbeeld laten we zien hoe je validatieruns plant met Airflow, dat zeer geschikt is voor het orkestreren van datapijplijnen.

Zo kun je een Airflow-DAG (Directed Acyclic Graph) instellen 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 een validatierun eenmaal per dag plant (@daily). 

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 voor datavalidatie met Great Expectations.

Begin klein en itereren

Begin met basiscontroles voor datakwaliteit en breid geleidelijk uit. Focus in het begin liever op eenvoudige expectations om onnodige complexiteit te vermijden—dat maakt integratie soepeler en troubleshooting eenvoudiger. Naarmate je de dataset beter leert kennen, kun je complexere validaties toevoegen.

Werk samen over teams heen

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

Automatiseer waar mogelijk

Automatiseer het proces waar dat kan om datavalidatie in je datapijplijnen op te nemen. 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 in Great Expectations configureert en valideert. Deze technieken helpen je om hoge datakwaliteit en transparantie in je workflows te behouden.

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

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