Great Expectations Tutorial: Daten mit Python validieren

Great Expectations einrichten

In diesem Tutorial lernst du, wie du GX Core, die Open-Source-Version von Great Expectations, verwendest, um ein Pandas-DataFrame zu validieren. Wir richten einen Context ein, registrieren eine Pandas-Datenquelle, definieren Expectations und validieren Daten-Batches.

Hinweis: Wir empfehlen, dem DataLab-Notebook zu folgen. Du kannst aber auch ein eigenes Python-Skript erstellen.

1. Great Expectations installieren

Voraussetzungen

• Python 3.9 bis 3.12 installiert.
• Um Konflikte zu vermeiden, installiere Great Expectations am besten in einer virtuellen Umgebung (Hinweis: Die Einrichtung virtueller Umgebungen sprengt den Rahmen dieses Artikels).
• Ein Beispieldatensatz.

Hinweis: Wenn du das bereitgestellte DataLab-Notebook verwendest, sind diese Voraussetzungen bereits erfüllt. Du kannst diesen Schritt überspringen.

Installiere GX mit folgendem pip-Befehl:

pip install great_expectations

Damit installierst du das Kernpaket inklusive aller benötigten Abhängigkeiten.

2. Data Context initialisieren

Great Expectations benötigt einen Data Context zur Verwaltung der Konfiguration. Wir verwenden einen ephemeren Context, damit keine Konfigurationen dauerhaft gespeichert werden.

import great_expectations as gx

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

Deine erste Datenvalidierungs-Suite erstellen

Nachdem GX eingerichtet ist, erstellen wir eine Validierungs-Suite.

1. Mit einer Datenquelle verbinden und ein Data Asset erstellen

Eine Datenquelle verbindet Great Expectations mit deinen Daten, ein Data Asset steht für einen konkreten Teil der Daten (z. B. eine Tabelle, ein DataFrame oder eine Datei).

In unserem Beispiel bereiten wir die Verbindung zu einem DataFrame namens inventory_parts_df vor. Der Beispieldatensatz ist im bereitgestellten DataLab enthalten und wird erzeugt, sobald wir den SQL-Block ausführen:

Wenn du nicht mit DataLab arbeitest, erstelle dein eigenes DataFrame mit Beispieldaten.

Jetzt legst du Datenquelle und Asset an:

# 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. Eine Batch-Definition hinzufügen

Eine Batch-Definition identifiziert und organisiert deine Daten für die Validierung. Hier fügen wir eine Batch-Definition hinzu, die das gesamte DataFrame umfasst:

# 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. Einen Batch abrufen

Ein Batch ist eine Datensammlung, die an eine Batch-Definition gekoppelt ist. Um zu validieren, rufst du den Batch ab und verknüpfst ihn mit deinem DataFrame, hier 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. Suite erstellen und Expectations definieren

Expectations sind Regeln zur Validierung deiner Daten. In diesem Beispiel definieren wir zwei einfache Regeln:

1. inventory_id darf nicht null sein.
2. part_num muss eindeutig sein.

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

Alle verfügbaren Expectations findest du in der Expectation Gallery. Probier gern noch ein paar weitere aus.

Nach dem Definieren der Expectations gibt GX die Suite-Konfiguration aus:

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

Die Suite enthält folgende Angaben:

1. Suite-Name und ID: Ein eindeutiger Name (inventory_parts_suite) und eine Kennung zur Verwaltung und Nachverfolgung.
2. Expectations: Jede Regel spezifiziert:
• Die Art der Prüfung (z. B. keine Nullwerte in einer Spalte oder eindeutige Einträge).
• Parameter wie die zu prüfende Spalte.
• Metadaten und eine eindeutige ID, um die Expectation leichter nachzuvollziehen und anzupassen.
3. Metadaten: Versionsinformationen zu Great Expectations für die Kompatibilität.
4. Notizen: Ein Platzhalter für erläuternde Kommentare zur Suite (optional).

Diese strukturierte Ausgabe dient zugleich als Dokumentation und wiederverwendbare Konfiguration für die Validierung – klar definiert, nachvollziehbar und bereit für den nächsten Einsatz.

5. Daten validieren

Zum Schluss validierst du den Batch gegen die definierten Expectations und bewertest die Ergebnisse.

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

Nach der Validierung liefert Great Expectations einen ausführlichen Bericht darüber, ob der Datensatz die Erwartungen erfüllt:

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

Der Bericht beschreibt die Datenqualität und hebt erfolgreiche sowie fehlgeschlagene Checks hervor. Eine kurze Einordnung der Ergebnisse:

Gesamtvalidierung: Das Ergebnis ist teilerfolgreich: 50% der Expectations wurden erfüllt, 50% nicht. Ein Fehlschlag weist auf ein Datenqualitätsproblem hin, das du prüfen solltest. Hier entspricht eine Spalte nicht der definierten Regel.

Expectation 1: inventory_id darf keine fehlenden Werte enthalten

• Ergebnis: Bestanden
• Erläuterung: In der Spalte inventory_id sind alle Werte vorhanden, es gibt keine Null- oder fehlenden Einträge. Das zeigt eine gute Datenvollständigkeit in dieser Spalte.

Expectation 2: part_num muss eindeutige Werte haben

• Ergebnis: Nicht bestanden
• Erläuterung: Die Spalte part_num enthält 97,98% Duplikate, nur wenige Werte sind eindeutig.
• Highlights:
• Beispielhafte Duplikate sind „3069b“ und „33291“.
• Das Tool zeigt auch die Häufigkeit dieser Duplikate und ihre Zeilenpositionen, sodass sich Probleme leichter finden und beheben lassen.

Natürlich handelt es sich um einen Beispieldatensatz. Wir haben bewusst eine bestandene und eine nicht bestandene Expectation aufgenommen, damit du beide Ergebnisse siehst.

Fertig! Du hast eine komplette Datenvalidierung von Ende zu Ende durchgeführt.

Great Expectations in Datenpipelines integrieren

Im Produktivbetrieb sollten Validierungen direkt in den Workflow eingebettet sein, um die Datenqualität in jeder Phase kontinuierlich zu überwachen.

In diesem Abschnitt zeigen wir, wie du Great Expectations in deine Datenpipelines integrierst.

Die folgenden Beispiele dienen als Orientierung. Je nach Tool sind zusätzliche Konfigurationen nötig. Prüfe die Dokumentation der jeweiligen Tools für die aktuelle Syntax.

Integration mit ETL-Tools

Die Integration von Great Expectations mit gängigen ETL-Tools wie Apache Airflow oder Prefect ist relativ unkompliziert. Wenn du Validierungsschritte direkt in ETL-Prozesse einbaust, erkennst und behebst du Datenprobleme in Echtzeit, bevor sie nachgelagerte Analysen beeinflussen.

Hier ein einfaches Beispiel, wie du Great Expectations mit Prefect verknüpfst, um die Datenvalidierung in einen automatisierten ETL-Workflow einzubetten:

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 diesem Beispiel definieren wir einen Prefect-Flow mit einer Aufgabe für die Great-Expectations-Validierung.

Die Funktion validate_data() lädt den Great-Expectations-Context, ruft den Daten-Batch ab und wendet die Expectation-Suite an.

Wenn die Daten die Kriterien nicht erfüllen, löst der Task einen Alarm aus, stoppt den Workflow und verhindert Fehler in nachgelagerten Schritten.

Kontinuierliche Datenvalidierung

Du kannst Validierungsjobs zeitgesteuert laufen lassen – etwa per Cron auf Unix-Systemen oder mit Diensten wie Apache Airflow. Im folgenden Beispiel planen wir Validierungsläufe mit Airflow, das sich gut zur Orchestrierung von Datenpipelines eignet.

So richtest du einen Airflow-DAG (Directed Acyclic Graph) ein, der Great-Expectations-Validierungen täglich ausführt:

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

Dieser DAG plant eine tägliche Ausführung (@daily).

Die Funktion run_validation() lädt den Great-Expectations-Context und führt die definierte Expectation-Suite gegen die Daten aus.

Best Practices für die Datenvalidierung mit Great Expectations

Für Skalierbarkeit und Effizienz gelten auch hier Best Practices – wie überall in der Datenarbeit.

Klein anfangen und iterieren

Starte mit grundlegenden Qualitätsprüfungen und erweitere Schritt für Schritt. Fokussiere dich zunächst auf Basiserwartungen, um die Integration schlank zu halten und die Fehlersuche zu vereinfachen. Mit wachsendem Verständnis des Datensatzes kannst du komplexere Validierungen ergänzen.

Teamübergreifend zusammenarbeiten

Datenqualität ist nicht nur ein technisches Thema. Lege Expectations gemeinsam mit Fachbereichen fest und stelle sicher, dass die Validierung zur Businesslogik und zu den Zielen passt. So erfüllen Daten ihren Zweck und Anforderungen aller Stakeholder.

Wo möglich automatisieren

Automatisiere Validierungen und integriere sie in deine Datenpipelines. Kontinuierliche, automatische Checks überwachen die Datenqualität ohne manuellen Aufwand – das steigert die Effizienz deutlich.

Fazit

Starke Leistung! Du hast gelernt, wie du Daten mit Great Expectations konfigurierst und validierst. Diese Techniken helfen dir, hohe Datenqualität und Transparenz in deinen Workflows zu sichern.

Wenn du weiter dazulernen möchtest, schau dir diese Ressourcen an:

• ETL and ELT in Python: Lerne, Daten effektiv zu transformieren und zu bewegen.
• Introduction to Data Quality: Entdecke die Grundlagen des Datenqualitätsmanagements.
• Cleaning Data in Python: Meistere Techniken zur Datenbereinigung für Genauigkeit und Konsistenz.
• Data Quality Dimensions Cheat Sheet: Eine praktische Übersicht zu Dimensionen der Datenqualität.