Валидация данных с Great Expectations

Введение

dbt tests (урок 5) отлично работают внутри dbt-pipeline. Но не все данные проходят через dbt: CSV-файлы от внешних лабораторий, API-данные, raw data в S3. Great Expectations (фреймворк) — open-source Python-фреймворк для валидации данных любого формата и из любого источника. В этом уроке мы разберём архитектуру GE и научимся создавать Expectation Suites.

Концепция: Expectations

В Great Expectations правило качества называется Expectation — декларативное утверждение о данных:

# "Я ожидаю, что значения в столбце payment_id уникальны"
expect_column_values_to_be_unique(column="payment_id")

# "Я ожидаю, что currency принимает только значения RUB, USD, EUR"
expect_column_values_to_be_in_set(column="currency", value_set=["RUB", "USD", "EUR"])

Каждый Expectation — это тест, который возвращает True/False с детальным отчётом.

Архитектура Great Expectations

Hands-On Lab: Quality Lab

Practice these concepts with an intentionally dirty dataset and Great Expectations:

cd labs/quality && cp .env.example .env && docker compose up -d --build

Open JupyterLab at http://localhost:28888 and complete Notebooks 02 and 03: Expectation Basics + Expectation Suite — write GE expectations against dirty data, build a full suite, and run a validation checkpoint.

Note: The lab uses GE 1.x API (current). This lesson references 0.18.x patterns for context — see the lab README for API differences.

Requirements: Docker Desktop with 4+ GB RAM allocated. See labs/quality/README.md for full setup.

expectation_suite:
  name: payments_quality_suite
  dataset: payments
  expectations:
    - type: expect_table_columns_to_match_set
      columns:
        - payment_id
        - order_id
        - amount
        - currency
        - payment_date
        - settlement_date
        - status
        - method

    - type: expect_column_values_to_be_unique
      column: payment_id

    - type: expect_column_values_to_not_be_null
      columns:
        - payment_id
        - order_id
        - amount
        - currency
        - payment_date
        - status

    - type: expect_column_values_to_be_in_set
      column: currency
      value_set: [RUB, USD, EUR]

    - type: expect_column_values_to_be_in_set
      column: status
      value_set: [pending, completed, failed, refunded]

    - type: expect_column_values_to_be_between
      column: amount
      min_value: 0.01
      max_value: 10000000

    - type: expect_column_pair_values_a_to_be_less_than_or_equal_to_b
      column_a: payment_date
      column_b: settlement_date

import great_expectations as gx

context = gx.get_context()

# Подключение к данным
datasource = context.sources.add_pandas("payments_source")
data_asset = datasource.add_csv_asset("payments", filepath="data/payments.csv")

# Создание suite
suite = context.add_expectation_suite("payments_quality_suite")

# Добавление expectations
validator = context.get_validator(
    batch_request=data_asset.build_batch_request(),
    expectation_suite_name="payments_quality_suite"
)

validator.expect_column_values_to_be_unique("payment_id")
validator.expect_column_values_to_not_be_null("payment_id")
validator.expect_column_values_to_be_in_set("currency", ["RUB", "USD", "EUR"])
validator.expect_column_values_to_be_between("amount", min_value=0.01, max_value=10000000)

# Сохранение suite
validator.save_expectation_suite()

# Создание checkpoint
checkpoint = context.add_or_update_checkpoint(
    name="payments_checkpoint",
    validations=[{
        "batch_request": data_asset.build_batch_request(),
        "expectation_suite_name": "payments_quality_suite"
    }]
)

# Запуск валидации
result = checkpoint.run()
print(f"Success: {result.success}")

context.build_data_docs()
context.open_data_docs()

GE vs dbt tests: когда что использовать

Сценарий: БиоГенезис

Сценарий: BioGenesis Lab (БиоГенезис Лаб)

БиоГенезис получает результаты клинических анализов от внешних лабораторий в CSV-файлах по email. Данные загружаются в PostgreSQL без валидации. Проблемы:

• Формат CSV различается между лабораториями (разные названия столбцов)
• Единицы измерения: одна лаборатория отправляет мг/дл, другая — ммоль/л
• Пропущенные значения не обозначены как NULL: используются пустые строки, “N/A”, ”-”

Great Expectations решает эти проблемы до загрузки в PostgreSQL:

expectations:
  - type: expect_table_columns_to_match_set
    columns: [patient_id, test_name, result_value, unit, lab_id, test_date]
  - type: expect_column_values_to_be_in_set
    column: unit
    value_set: [mg/dL, mmol/L, g/L, U/L]
  - type: expect_column_values_to_not_be_null
    columns: [patient_id, test_name, result_value]

Итоги

• Great Expectations — Python-фреймворк для валидации любых данных (CSV, API, S3, DB)
• Expectation — декларативное правило качества: expect_column_values_to_be_unique, expect_column_values_to_be_in_set
• Expectation Suite — набор правил для одного датасета
• Checkpoint — оркестрация: запуск suite на данных с генерацией отчёта
• Data Docs — автоматическая HTML-документация результатов
• GE и dbt tests комплементарны: dbt для warehouse, GE для raw data

В следующем уроке мы завершим модуль темой Data Observability — непрерывного мониторинга здоровья data pipeline, включая freshness SLA, incident response и интеграцию с CDC-системами.