Руководство по Great Expectations: Проверка данных с помощью Python

Настройка Great Expectations

В этом учебнике вы узнаете, как использовать GX Core, open-source версию Great Expectations, для валидации Pandas DataFrame. Мы рассмотрим настройку контекста, регистрацию источника данных Pandas, определение ожиданий и проверку пакетов данных.

Примечание: Мы рекомендуем вам следовать за DataLab notebook, но вы также можете создать свой собственный скрипт на Python.

1. Установка Great Expectations

Предварительные требования

• Python 3.9 до 3.12 установлен.
• Чтобы избежать конфликтов, настоятельно рекомендуется установить Great Expectations в виртуальной среде (отказ от ответственности: настройка виртуальных сред выходит за рамки данной статьи).
• Пример набора данных.

Примечание: Если вы используете предоставленный ноутбук DataLab, эти предварительные требования уже выполнены. Можете смело их пропустить.

Используйте следующую команду для установки GX через pip:

pip install great_expectations

Эта команда устанавливает основной пакет и все необходимые зависимости.

2. Инициализация контекста данных

Great Expectations требует контекста данных для управления конфигурациями. Мы используем эфемерный контекст данных, чтобы избежать сохранения конфигураций.

import great_expectations as gx

# Получение эфемерного контекста данных
context = gx.get_context()
assert type(context).__name__ == "EphemeralDataContext"

Создание вашего первого набора проверки данных

Теперь, когда GX настроен, давайте создадим набор проверки данных.

1. Подключение к источнику данных и создание актива данных

Источник данных соединяет Great Expectations с вашими данными, в то время как актив данных представляет собой конкретный поднабор данных (например, таблицу, DataFrame или файл).

В этом случае мы подготовим все для подключения к DataFrame с именем inventory_parts_df. Образец набора данных доступен в предоставленном DataLab и создается после выполнения SQL-блока:

Если вы не используете DataLab, создайте свой собственный DataFrame с образцовыми данными.

Теперь создайте свой источник данных и актив:

# Добавление источника данных Pandas
data_source = context.data_sources.add_pandas(name="inventory_parts")
# Добавление актива в источник данных
data_asset = data_source.add_dataframe_asset(name="inventory_parts_asset")

2. Добавление определения пакета

Определение пакета идентифицирует и организует ваши данные для проверки. Здесь мы добавляем определение пакета, которое охватывает весь DataFrame:

# Определите имя определения пакета
batch_definition_name = "inventory_parts_batch"
# Добавить определение пакета
batch_definition = data_asset.add_batch_definition_whole_dataframe(batch_definition_name)
assert batch_definition.name == batch_definition_name

3. Извлечение пакета

Пакет — это коллекция данных, связанная с определением пакета. Для проверки данных вам нужно извлечь и связать пакет с вашим DataFrame, в данном случае inventory_parts_df:

# Определите параметры пакета
batch_parameters = {"dataframe": inventory_parts_df}
# Извлеките пакет
batch = batch_definition.get_batch(batch_parameters=batch_parameters)

4. Создание набора и определение ожиданий

Ожидания — это правила для проверки данных. В этом примере мы определим следующие простые ожидания:

1. Убедитесь, что значения inventory_id не равны null.
2. Убедитесь, что значения part_num уникальны.

# Создание набора ожиданий
expectation_suite_name = "inventory_parts_suite"
suite = gx.ExpectationSuite(name=expectation_suite_name)
# Добавление ожиданий
suite.add_expectation(
    gx.expectations.ExpectColumnValuesToNotBeNull(column="inventory_id")
)
suite.add_expectation(
    gx.expectations.ExpectColumnValuesToBeUnique(column="part_num")
)
# Добавление набора ожиданий в контекст
context.suites.add(suite)

Вы можете изучить все доступные ожидания в Галерее ожиданий. Мы рекомендуем вам добавить еще несколько!

После определения ожиданий, GX выводит настройки набора ожиданий:

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

В набор включены следующие детали:

1. Имя и идентификатор набора: Уникальное имя (inventory_parts_suite) и идентификатор для отслеживания и управления набором.
2. Ожидания: Каждое правило указывает:
• Тип проверки (например, убедиться, что столбец не имеет пустых значений или уникальных записей).
• Параметры, такие как проверяемый столбец.
• Метаданные и уникальный идентификатор для каждого ожидания, облегчающие отслеживание и настройку.
3. Метаданные: Информация о версии Great Expectations, обеспечивающая совместимость с инструментом.
4. Примечания: Заполнитель для добавления описательных комментариев о наборе (по желанию).

Этот структурированный вывод действует как документация и повторно используемая конфигурация для проверки вашего набора данных, чтобы ваши ожидания были четко определены, прослеживаемы и готовы к будущему использованию.

5. Проверка данных

Наконец, выполните проверку партии по заданным ожиданиям и оцените результаты.

# Проверка данных по набору
validation_results = batch.validate(suite)
# Оценка результатов
print(validation_results)

После запуска проверки Great Expectations предоставляет подробный отчет о том, соответствует ли набор данных заданным ожиданиям:

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

Этот отчет подробно описывает качество ваших данных, выделяя успехи и неудачи. Вот упрощенное объяснение результатов:

Общая проверка: Результат проверки был частично успешным: 50% ожиданий были выполнены, а 50% не выполнены. Неудачное ожидание указывает на проблему качества данных, требующую внимания. В этом случае один столбец не соответствовал определенному правилу.

Ожидание 1: inventory_id не должен содержать отсутствующих значений

• Результат: Пройдено
• Объяснение: Каждое значение в столбце inventory_id присутствует, без нулевых или отсутствующих записей. Это указывает на хорошую полноту данных для этого столбца.

Ожидание 2: part_num должен иметь уникальные значения

• Результат: Сбой
• Объяснение: Столбец part_num содержит 97.98% дублирующихся значений, что означает, что только несколько значений являются уникальными.
• Основные моменты:
• Примеры дублирующихся значений включают “3069b” и “33291”.
• Инструмент также показывает, как часто эти дубликаты появляются и их позиции строк, что облегчает обнаружение и исправление проблем.

Конечно, это всего лишь образец набора данных, и мы специально включили проходные и не проходные ожидания, чтобы вы могли увидеть оба результата проверки.

Вот и все! Вы успешно запустили проверку данных от начала до конца.

Интеграция Great Expectations в потоки данных

В рабочей среде проверки должны быть встроены непосредственно в рабочий процесс для непрерывного мониторинга качества данных на каждом этапе.

В этом разделе мы обсудим, как вы можете интегрировать Great Expectations в ваши потоки данных.

Это примеры, чтобы дать вам представление, и для выполнения дополнительных настроек, не включенных здесь, может потребоваться. Ознакомьтесь с документацией каждого инструмента для актуального синтаксиса!

Интеграция с ETL-инструментами

Интеграция Great Expectations с популярными ETL-инструментами, такими как Apache Airflow или Prefect, довольно проста. Встраивание шагов валидации непосредственно в процессы ETL позволит вам обнаружить и исправить проблемы с данными в реальном времени до их влияния на последующий анализ.

Давайте рассмотрим простой пример интеграции Great Expectations с Prefect для выполнения проверки данных в рамках автоматизированного рабочего процесса ETL:

from prefect import task, Flow
import great_expectations as ge
# Определите задачу для выполнения валидации Great Expectations
@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])
	    
    # Проверьте результаты проверки и выведите предупреждение, если проверка не пройдена
    if not results["success"]:
        raise ValueError("Data validation failed!")
# Задайте ваш поток ETL
with Flow("ETL_with_GE_Validation") as flow:
    validation = validate_data()
# Выполните поток
flow.run()

В этом примере мы определяем поток Prefect с задачей для выполнения валидации Great Expectations.

Задача validate_data() загружает контекст Great Expectations, извлекает пакет данных и применяет набор ожиданий.

Если данные не соответствуют критериям валидации, задача генерирует предупреждение, останавливая рабочий процесс и предотвращая появление ошибок вниз по потоку.

Непрерывная валидация данных

Вы можете планировать задания по валидации с использованием различных инструментов, таких как cron-задания в системах на основе Unix или управляемые сервисы, например, Apache Airflow. В этом примере мы продемонстрируем, как планировать запуск валидации с использованием Airflow, который отлично подходит для оркестрации потоков данных.

Вот как вы можете настроить DAG Airflow (Directed Acyclic Graph) для запуска валидации Great Expectations ежедневно:

from airflow import DAG
from airflow.operators.python_operator import PythonOperator
from datetime import datetime
import great_expectations as ge
# Определите DAG и установите расписание на ежедневное выполнение
default_args = {
	  'owner': 'airflow',
	  'start_date': datetime(2024, 1, 1),
	  'retries': 1,
}
dag = DAG(
      'great_expectations_validation',
	default_args=default_args,
	schedule_interval='@daily',  # Запуск один раз в день
)
# Определите функцию для выполнения проверки
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
# Настройте задачу в Airflow
validation_task = PythonOperator(
      task_id='run_great_expectations_validation',
      python_callable=run_validation,
      dag=dag,
)
# Установите задачу в DAG
validation_task

В этом примере мы определяем DAG, который планирует выполнение проверки один раз в день (@daily).

Функция run_validation() выполняет проверку, загружая контекст Great Expectations и запуская определенный набор ожиданий для данных.

Лучшие практики для валидации данных с Great Expectations

Всегда лучше придерживаться лучших практик для масштабируемости и эффективности, и это не исключение и для валидации данных с Great Expectations.

Начните с малого и итерируйте.

Начните с проверки основных данных качества и постепенно расширяйтесь. Лучше сосредоточиться на базовых ожиданиях изначально, так как это поможет избежать усложнения процесса, что облегчает интеграцию и упрощает устранение неполадок. По мере улучшения понимания набора данных можно добавлять более сложные проверки.

Сотрудничество между командами

Качество данных — это не только техническое вопрос. Сотрудничайте с бизнес-командами для определения ожиданий и обеспечения того, что внедренная проверка соответствует основной бизнес-логике и целям. Такой кросс-функциональный подход гарантирует, что данные служат своей предназначенной цели и соответствуют требованиям всех заинтересованных сторон.

Автоматизируйте, где это возможно

Автоматизируйте процесс там, где это возможно, чтобы интегрировать проверку данных в потоки данных. Интеграция автоматических проверок позволяет непрерывно контролировать качество данных без ручного вмешательства, что значительно повышает эффективность.

Заключение

Отличная работа! Вы научились настраивать и проверять данные в Great Expectations. Эти техники помогут поддерживать высокое качество данных и прозрачность в ваших рабочих процессах.

Чтобы продолжить развивать свои навыки, ознакомьтесь с этими ресурсами:

• ETL и ELT на Python: Узнайте, как эффективно преобразовывать и перемещать данные.
• Введение в качество данных: Изучите основы управления качеством данных.
• Очистка данных на Python: Освойте техники очистки данных для обеспечения точности и согласованности.
• Шпаргалка по измерениям качества данных: Удобное руководство по измерениям качества данных.