Samouczek Great Expectations: walidacja danych w Pythonie

Konfiguracja Great Expectations

W tym samouczku nauczysz się korzystać z GX Core, otwartoźródłowej wersji Great Expectations, do walidacji obiektu Pandas DataFrame. Przejdziemy przez konfigurację kontekstu, rejestrację źródła danych Pandas, definiowanie oczekiwań i walidację wsadów danych.

Uwaga: Zalecamy, byś śledził kroki w notatniku DataLab, ale możesz też stworzyć własny skrypt Pythona.

1. Instalacja Great Expectations

Wymagania wstępne

• Zainstalowany Python w wersji 3.9–3.12.
• Aby uniknąć konfliktów, zdecydowanie zaleca się instalację Great Expectations w środowisku wirtualnym (zastrzeżenie: konfiguracja środowisk wirtualnych wykracza poza zakres tego artykułu).
• Przykładowy zbiór danych.

Uwaga: Jeśli używasz dostarczonego notatnika DataLab, te wymagania są już spełnione. Możesz je pominąć.

Użyj poniższego polecenia, aby zainstalować GX przez pip:

pip install great_expectations

To polecenie instaluje główny pakiet i wszystkie niezbędne zależności.

2. Inicjalizacja kontekstu danych

Great Expectations wymaga kontekstu danych do zarządzania konfiguracjami. Używamy ulotnego (ephemeral) kontekstu danych, aby nie zapisywać konfiguracji na stałe.

import great_expectations as gx

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

Tworzenie pierwszego zestawu walidacji danych

Skoro GX jest już skonfigurowane, utwórzmy zestaw walidacji danych.

1. Połączenie ze źródłem danych i utworzenie zasobu danych

Źródło danych łączy Great Expectations z twoimi danymi, a zasób danych reprezentuje określony podzbiór danych (np. tabelę, DataFrame lub plik).

W tym przypadku przygotujemy wszystko do połączenia z DataFrame o nazwie inventory_parts_df. Przykładowy zbiór danych jest dostępny w dostarczonym DataLab i zostaje utworzony po uruchomieniu bloku SQL:

Jeśli nie korzystasz z DataLab, utwórz własny DataFrame z przykładowymi danymi.

Teraz utwórz swoje źródło i zasób danych:

# 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. Dodanie definicji wsadu

Definicja wsadu identyfikuje i organizuje twoje dane do walidacji. Tutaj dodamy definicję wsadu obejmującą cały 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. Pobranie wsadu

Wsad to kolekcja danych powiązana z definicją wsadu. Aby zwalidować dane, musisz pobrać wsad i powiązać go ze swoim DataFrame, w tym przypadku 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. Utworzenie zestawu i zdefiniowanie oczekiwań

Oczekiwania to reguły walidacji danych. W tym przykładzie zdefiniujemy następujące proste oczekiwania:

1. Zapewnienie, że wartości inventory_id nie są puste (null).
2. Zapewnienie, że wartości part_num są unikalne.

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

Wszystkie dostępne oczekiwania możesz przejrzeć w Expectation Gallery. Zachęcamy, byś dodał kilka więcej!

Po zdefiniowaniu oczekiwań GX wyświetla konfigurację zestawu oczekiwań:

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

Zestaw zawiera następujące szczegóły:

1. Nazwa i ID zestawu: Unikalna nazwa (inventory_parts_suite) i identyfikator do śledzenia i zarządzania zestawem.
2. Oczekiwania: Każda reguła określa:
• Typ sprawdzenia (np. upewnienie się, że kolumna nie ma wartości null lub że zawiera unikalne wpisy).
• Parametry, takie jak kolumna poddawana walidacji.
• Metadane i unikalny ID dla każdego oczekiwania, co ułatwia śledzenie i dostosowanie.
3. Metadane: Informacje o wersji Great Expectations, zapewniające zgodność z narzędziem.
4. Notatki: Miejsce na dodanie opisowych komentarzy o zestawie (opcjonalne).

Ta ustrukturyzowana odpowiedź pełni rolę zarówno dokumentacji, jak i wielokrotnego użytku konfiguracji do walidacji twojego zbioru danych — twoje oczekiwania są jasno zdefiniowane, możliwe do prześledzenia i gotowe do ponownego użycia.

5. Walidacja danych

Na koniec zwaliduj wsad względem zdefiniowanych oczekiwań i oceń wyniki.

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

Po uruchomieniu walidacji Great Expectations dostarcza szczegółowy raport o tym, czy zbiór danych spełnia zdefiniowane oczekiwania:

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

Ten raport opisuje jakość twoich danych, wskazując sukcesy i niepowodzenia. Oto uproszczone wyjaśnienie wyników:

Walidacja ogólna: Wynik walidacji był częściowo pomyślny: 50% oczekiwań przeszło, a 50% nie. Niepowodzenie oznacza problem z jakością danych, który wymaga uwagi. W tym przypadku jedna kolumna nie spełniła zdefiniowanej reguły.

Oczekiwanie 1: inventory_id nie powinno mieć brakujących wartości

• Wynik: Zaliczono
• Wyjaśnienie: Każda wartość w kolumnie inventory_id jest obecna — brak nulli czy braków. To oznacza dobrą kompletność danych dla tej kolumny.

Oczekiwanie 2: part_num powinno mieć unikalne wartości

• Wynik: Nie zaliczono
• Wyjaśnienie: Kolumna part_num zawiera 97,98% wartości zduplikowanych, czyli tylko niewielka część wartości jest unikalna.
• Najważniejsze:
• Przykładowe duplikaty to „3069b” i „33291”.
• Narzędzie pokazuje też częstość występowania tych duplikatów i ich pozycje w wierszach, co ułatwia znalezienie i naprawę problemów.

Oczywiście to tylko przykładowy zbiór danych — celowo uwzględniliśmy jedno zaliczone i jedno niezaliczone oczekiwanie, żebyś zobaczył oba wyniki walidacji.

I to wszystko! Pomyślnie uruchomiłeś walidacje danych end‑to‑end.

Integracja Great Expectations z potokami danych

W środowisku produkcyjnym walidacje muszą być wbudowane bezpośrednio w przepływ pracy, aby stale monitorować jakość danych na każdym etapie. 

W tej sekcji omówimy, jak możesz zintegrować Great Expectations ze swoimi potokami danych.

To przykłady poglądowe — mogą być potrzebne dodatkowe konfiguracje niewymienione tutaj. Sprawdź dokumentację każdego narzędzia, by mieć aktualną składnię!

Integracja z narzędziami ETL

Integracja Great Expectations z popularnymi narzędziami ETL, takimi jak Apache Airflow czy Prefect, jest stosunkowo prosta. Wbudowanie kroków walidacji bezpośrednio w procesy ETL pozwala wychwytywać i rozwiązywać problemy z danymi w czasie rzeczywistym, zanim wpłyną na analizy w dalszej części.

Przejdźmy przez prosty przykład integracji Great Expectations z Prefect, aby uruchomić walidację danych jako część zautomatyzowanego przepływu ETL:

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

W tym przykładzie definiujemy przepływ Prefect z zadaniem uruchamiającym walidację w Great Expectations.

Zadanie validate_data() ładuje kontekst Great Expectations, pobiera wsad danych i stosuje zestaw oczekiwań. 

Jeśli dane nie spełniają kryteriów walidacji, zadanie zgłasza alert, zatrzymując przepływ pracy i zapobiegając błędom w dalszych etapach.

Ciągła walidacja danych

Możesz harmonogramować uruchamianie walidacji za pomocą różnych narzędzi, takich jak cron w systemach Unix lub usługi zarządzane, np. Apache Airflow. W tym przykładzie pokażemy, jak zaplanować uruchomienia walidacji w Airflow, które świetnie nadaje się do orkiestracji potoków danych.

Oto jak możesz skonfigurować DAG Airflow (Directed Acyclic Graph), by codziennie uruchamiać walidacje Great Expectations:

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

W tym przykładzie definiujemy DAG, który planuje uruchomienie walidacji raz dziennie (@daily). 

Funkcja run_validation() wykonuje walidację, ładując kontekst Great Expectations i uruchamiając zdefiniowany zestaw oczekiwań na danych.

Najlepsze praktyki walidacji danych z Great Expectations

Dla skalowalności i efektywności zawsze warto trzymać się dobrych praktyk — nie inaczej jest w przypadku walidacji danych z Great Expectations.

Zacznij od małych kroków i iteruj

Zacznij od podstawowych kontroli jakości danych i stopniowo je rozszerzaj. Lepiej na początku skupić się na podstawowych oczekiwaniach, by nie komplikować procesu — to ułatwia integrację i rozwiązywanie problemów. Gdy lepiej poznasz zbiór danych, możesz dodawać bardziej złożone walidacje.

Współpracuj między zespołami

Jakość danych to nie tylko kwestia techniczna. Współpracuj z zespołami biznesowymi, aby zdefiniować oczekiwania i upewnić się, że wdrożona walidacja odpowiada logice biznesowej i celom. Takie podejście międzyfunkcyjne gwarantuje, że dane służą zamierzonemu celowi i spełniają wymagania wszystkich interesariuszy.

Automatyzuj, gdzie to możliwe

Automatyzuj proces wszędzie tam, gdzie to wykonalne, aby włączyć walidację danych do potoków. Zautomatyzowane kontrole walidacyjne umożliwiają ciągłe monitorowanie jakości danych bez interwencji ręcznej, co znacząco zwiększa efektywność.

Podsumowanie

Świetna robota! Nauczyłeś się konfigurować i walidować dane w Great Expectations. Te techniki pomogą utrzymać wysoką jakość danych i przejrzystość twoich przepływów pracy.

Aby dalej rozwijać umiejętności, sprawdź te zasoby:

• ETL i ELT w Pythonie: Naucz się skutecznie przekształcać i przenosić dane.
• Wprowadzenie do jakości danych: Poznaj podstawy zarządzania jakością danych.
• Czyszczenie danych w Pythonie: Opanuj techniki czyszczenia danych, aby zapewnić dokładność i spójność.
• Ściąga: wymiary jakości danych: Poręczny przewodnik po wymiarach jakości danych.