Track
Инженер данных на Python
40 ч
Настройка Great Expectations
В этом руководстве вы узнаете, как использовать GX Core, open-source версию Great Expectations, для проверки DataFrame в Pandas. Мы пройдём настройку контекста, регистрацию источника данных Pandas, определение expectations и валидацию пакетов данных.
Примечание: Рекомендуем работать параллельно в ноутбуке DataLab, но вы также можете создать собственный скрипт на 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
# Get the Ephemeral Data Context
context = gx.get_context()
assert type(context).__name__ == "EphemeralDataContext"Создание первого набора правил валидации
Теперь, когда GX настроен, создадим набор правил для проверки данных.
1. Подключение к источнику данных и создание data asset
Источник данных подключает Great Expectations к вашим данным, а data asset представляет конкретное подмножество данных (например, таблицу, DataFrame или файл).
В нашем случае мы подготовим всё для подключения к DataFrame с именем inventory_parts_df. Пример набора данных доступен в предоставленном DataLab, и он создаётся после запуска SQL-блока:
Если вы не используете DataLab, создайте собственный DataFrame с тестовыми данными.
Теперь создайте источник и asset:
# 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. Добавление определения пакета (batch definition)
Batch definition определяет и организует ваши данные для валидации. Здесь мы добавим определение пакета, охватывающее весь 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_name3. Получение пакета
Пакет (batch) — это набор данных, связанный с batch definition. Чтобы выполнить проверку, нужно получить пакет и связать его с вашим DataFrame, в нашем случае с 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. Создание набора и определение expectations
Expectations — это правила для проверки данных. В этом примере мы зададим следующие простые правила:
- Гарантировать, что значения
inventory_idне равны null. - Гарантировать, что значения
part_numуникальны.
# 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)Вы можете изучить все доступные правила в Expectation Gallery. Добавьте ещё несколько — это полезно!
После определения правил GX выводит конфигурацию набора expectations:
{
"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
}Набор включает следующие детали:
- Имя набора и ID: Уникальное имя (
inventory_parts_suite) и идентификатор для отслеживания и управления набором. - Expectations: Каждое правило указывает:
- Тип проверки (например, отсутствие null в столбце или уникальность значений).
- Параметры, такие как проверяемый столбец.
- Метаданные и уникальный ID каждого правила для удобного отслеживания и настройки.
- Метаданные: Информация о версии Great Expectations для обеспечения совместимости.
- Заметки: Поле для добавления описательных комментариев о наборе (необязательно).
Эта структурированная конфигурация служит и документацией, и многоразовой настройкой для проверки вашего набора данных, чтобы ваши expectations были чётко определены, отслеживаемы и готовы к повторному использованию.
5. Валидация данных
Наконец, проверьте пакет на соответствие заданным правилам и оцените результаты.
# Validate the Data Against the Suite
validation_results = batch.validate(suite)
# Evaluate the Results
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% — нет. Неуспешная проверка указывает на проблему качества данных, требующую внимания. В данном случае один столбец не соответствует заданному правилу.
Expectation 1: inventory_id не должен содержать пропусков
- Результат: успешно
- Пояснение: В столбце
inventory_idприсутствуют все значения, нет null или пропусков. Это говорит о хорошей полноте данных в этом столбце.
Expectation 2: part_num должен содержать уникальные значения
- Результат: неуспешно
- Пояснение: В столбце
part_num97,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
# 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()В этом примере мы определяем поток Prefect с задачей для запуска проверки Great Expectations.
Задача validate_data() загружает контекст Great Expectations, получает пакет данных и применяет набор правил.
Если данные не соответствуют критериям проверки, задача генерирует оповещение, останавливая рабочий процесс и предотвращая ошибки далее по цепочке.
Непрерывная валидация данных
Вы можете планировать задания по валидации с помощью различных средств — от cron в Unix-системах до управляемых сервисов вроде Apache Airflow. В качестве примера покажем, как расписать проверки в Airflow, который хорошо подходит для оркестрации конвейеров данных.
Вот как настроить DAG (ориентированный ацикличный граф) в Airflow для ежедневного запуска проверок 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В этом примере мы определяем DAG, который запускает проверку раз в день (@daily).
Функция run_validation() выполняет проверку, загружая контекст Great Expectations и прогоняя определённый набор правил на данных.
Лучшие практики проверки данных с Great Expectations
Следование лучшим практикам — залог масштабируемости и эффективности; для Great Expectations это также справедливо.
Начинайте с малого и развивайте решение итеративно
Начните с базовых проверок качества и постепенно расширяйте их. Лучше сосредоточиться на простых правилах сначала, чтобы не усложнять процесс — так интеграция пройдёт гладко, а отладка будет проще. По мере углубления понимания набора данных добавляйте более сложные проверки.
Сотрудничайте между командами
Качество данных — это не только техническая задача. Сотрудничайте с бизнес-командами, чтобы определить правила и убедиться, что реализованная проверка соответствует бизнес-логике и целям. Такой кросс-функциональный подход гарантирует, что данные служат своей цели и отвечают требованиям всех заинтересованных сторон.
Автоматизируйте, где это возможно
Автоматизируйте процесс там, где это возможно, чтобы встроить проверки в конвейеры данных. Автоматические проверки обеспечивают непрерывный мониторинг качества без ручного вмешательства, заметно повышая эффективность.
Заключение
Отличная работа! Вы узнали, как настраивать и проверять данные в Great Expectations. Эти техники помогут поддерживать высокое качество данных и прозрачность ваших процессов.
Чтобы продолжить обучение, посмотрите эти материалы:
- ETL и ELT на Python: Научитесь эффективно преобразовывать и переносить данные.
- Введение в качество данных: Изучите основы управления качеством данных.
- Очистка данных на Python: Освойте приёмы очистки данных для обеспечения точности и согласованности.
- Шпаргалка по измерениям качества данных: Удобное руководство по измерениям качества данных.