Тестирование этапов конвейера с помощью утверждений
Добавляйте на каждом этапе проверки количества строк, утверждения об отсутствии пустых значений и проверки ожидаемых столбцов, чтобы ошибки выявлялись немедленно.
«Тестирование этапов конвейера с помощью утверждений» — бесплатный урок Pandas & NumPy Academy на CoddyKit. Это урок 3 из 4. Ты можешь прочитать весь урок бесплатно ниже — а потом практиковать его прямо в браузере с встроенным редактором кода и ИИ-репетитором 24/7. Это часть пути обучения Pandas & NumPy Academy, и твой прогресс синхронизируется между веб-версией и приложением CoddyKit. Курс Pandas & NumPy Academy содержит 4 уроков всего.
Зачем тестировать шаги конвейера?
Конвейер обработки данных, который работает без ошибок, всё равно может создавать незаметно неправильный результат: строки могут удаляться неправильным фильтром, столбец может умножаться на неверный коэффициент, а объединение — дублировать строки, если ключ объединения не был уникальным. Единственный способ обнаружить такие незаметные сбои — встроить утверждения, проверяющие ожидаемые свойства данных на каждом этапе конвейера. Утверждения превращают логические ошибки в явные и сразу заметные ошибки.
import pandas as pd
import numpy as np
# A transform that looks correct but has a bug:
def compute_revenue_buggy(df):
# BUG: unit_price should be multiplied, not added
df['revenue'] = df['quantity'] + df['unit_price']
return df
# Without assertions, this runs silently with wrong numbers.Проверки количества строк
После каждого шага фильтрации проверяйте, что количество получившихся строк находится в ожидаемом диапазоне. Слишком малое количество строк означает, что фильтр оказался чрезмерно строгим; слишком большое — что объединение продублировало записи. Задавайте ожидаемый диапазон как долю от входных данных: например, при корректных данных шаг удаления пустых значений не должен удалять более 30 % строк. Эта проверка выявляет неожиданные изменения качества данных в исходных источниках.
def drop_nulls_guarded(df, required_cols, max_drop_fraction=0.3):
before = len(df)
result = df.dropna(subset=required_cols)
after = len(result)
drop_fraction = (before - after) / before
assert drop_fraction <= max_drop_fraction, \
f'dropna removed {drop_fraction:.1%} of rows (limit {max_drop_fraction:.1%})'
return resultПроверка наличия столбцов
Проверяйте, что все ожидаемые выходные столбцы существуют после каждой функции преобразования. Если на шаге переименования случайно указан неправильный ключ, результирующий столбец будет отсутствовать, а ошибка проявится гораздо позже, когда другой шаг попытается его использовать. Проверка сразу после преобразования обнаруживает проблему в месте её возникновения, а не через три шага в виде непонятной ошибки KeyError.
def compute_revenue(df):
df = df.assign(revenue=lambda d: d['quantity'] * d['unit_price'])
# Guard: check that the new column exists and is non-null
assert 'revenue' in df.columns, 'revenue column not created'
assert df['revenue'].notna().all(), 'revenue has unexpected NaNs'
return dfПроверка диапазона значений
После вычисления столбца выручки проверяйте, что значения неотрицательны. После разбора дат проверяйте, что они относятся к ожидаемому диапазону лет. После кодирования категорий проверяйте, что неожиданные коды не появились. Каждая такая проверка является контрактом данных: она документирует ожидаемое поведение и помогает рано обнаружить нарушения. Записывайте их в виде проверок, а не операторов печати, чтобы при автоматическом запуске конвейера они вызывали ошибки.
def validate_computed_columns(df):
assert (df['revenue'] >= 0).all(), \
f'Negative revenue: {df[df["revenue"] < 0]["revenue"].head().tolist()}'
assert (df['quantity'] > 0).all(), \
'Non-positive quantity found'
assert df['revenue'].between(0, 1_000_000).all(), \
'Revenue out of plausible range'
print('Value range checks passed.')Защита от дублирования после объединений
Распространённая ошибка в данных — объединение «многие ко многим», случайно увеличивающее количество строк. После каждого pd.merge() проверяйте, что количество строк соответствует ожиданиям — обычно оно не должно превышать количество строк в левом DataFrame при левом объединении. Также проверяйте уникальность столбца-ключа, если она должна соблюдаться, чтобы сразу обнаружить случайное декартово объединение.
def safe_merge(left, right, on, how='left'):
before = len(left)
result = left.merge(right, on=on, how=how)
after = len(result)
if how == 'left':
assert after == before, \
f'Left join increased rows from {before} to {after} — check key uniqueness in right df'
return resultПроверка отсутствующих значений после важных шагов
Некоторые столбцы ни на одном этапе конвейера не должны содержать пустых значений. После каждого шага, который может случайно добавить пустые значения, проверяйте df['key_col'].notna().all() — например, после объединения, при котором не удалось сопоставить некоторые строки и в объединённых столбцах появились NaN, или после вызова map(), возвращающего NaN для несопоставленных значений. Делайте эти проверки двумя последними строками каждой функции преобразования, затрагивающей такие столбцы.
NOT_NULL_AFTER_TRANSFORM = ['order_id', 'revenue', 'category']
def post_transform_checks(df):
for col in NOT_NULL_AFTER_TRANSFORM:
null_count = df[col].isna().sum()
assert null_count == 0, \
f'{col}: {null_count} unexpected NaN values after transform'
print('Null checks passed.')
return dfСоздание набора тестов для шагов конвейера
Пишите модульные тесты для каждой функции конвейера, используя небольшие вручную созданные DataFrame, изолирующие проверяемую логику. Каждый тест должен выполнять три действия: подготавливать минимальные входные данные, вызывать функцию и проверять свойства результата. Для простых конвейеров достаточно встроенного в Python оператора assert; в крупных проектах используйте pytest, чтобы автоматически запускать все тесты перед развёртыванием.
def test_compute_revenue():
test_df = pd.DataFrame({
'quantity': [2, 3],
'unit_price': [10.0, 5.0]
})
result = compute_revenue(test_df)
assert 'revenue' in result.columns
assert result['revenue'].tolist() == [20.0, 15.0]
assert result['revenue'].dtype == float
print('test_compute_revenue PASSED')
test_compute_revenue()Проверка граничных случаев
Хорошие тесты охватывают не только штатный сценарий, но и граничные случаи: входные данные, полностью состоящие из пустых значений, пустой DataFrame, DataFrame с одной строкой и столбцы с экстремальными значениями. Пустой DataFrame должен возвращать пустой DataFrame, а не ошибку. DataFrame с одной строкой должен давать правильный результат. Проверяйте эти случаи явно, чтобы в рабочей среде конвейер корректно обрабатывал необычные входные данные.
def test_empty_df():
empty = pd.DataFrame({'quantity': [], 'unit_price': []})
result = compute_revenue(empty)
assert len(result) == 0, 'Empty input should produce empty output'
assert 'revenue' in result.columns, 'Revenue column should still be created'
print('test_empty_df PASSED')
def test_single_row():
single = pd.DataFrame({'quantity': [1], 'unit_price': [99.0]})
result = compute_revenue(single)
assert result['revenue'].iloc[0] == 99.0
print('test_single_row PASSED')
test_empty_df()
test_single_row()Интеграционный тест: весь конвейер на примере данных
Помимо модульных тестов отдельных функций напишите интеграционный тест, который запускает весь конвейер на небольшой репрезентативной выборке реальных данных. Проверьте, что результат содержит ожидаемое количество столбцов, столбец-ключ уникален, а общая выручка находится в правдоподобном диапазоне. Такая сквозная проверка выявляет ошибки во взаимодействии шагов, которые не обнаруживаются модульными тестами.
def integration_test(config):
raw = extract(config)
clean = transform(raw, config)
assert set(config['required_cols']).issubset(set(clean.columns))
assert clean['order_id'].is_unique
assert (clean['revenue'] >= 0).all()
assert len(clean) > 0
print(f'Integration test PASSED. Output: {clean.shape}')
integration_test(CONFIG)Непрерывное тестирование с pytest
По мере роста конвейера собирайте все тесты в каталоге tests/ и запускайте их из командной строки с помощью pytest. Файл conftest.py может создавать общие фикстуры, например примеры DataFrames. Интегрируйте pytest в свой конвейер CI/CD, чтобы при каждом изменении кода все тесты автоматически запускались до развёртывания. Неудачный тест блокирует развёртывание и не позволяет неработающему коду попасть в рабочую среду.
# tests/test_transform.py — example structure
# import pytest, pandas as pd
# from pipeline.transform import compute_revenue, drop_nulls
# @pytest.fixture
# def sample_df():
# return pd.DataFrame({'quantity': [2, 3], 'unit_price': [10.0, 5.0]})
# def test_revenue(sample_df):
# result = compute_revenue(sample_df)
# assert result['revenue'].tolist() == [20.0, 15.0]
print('Run: pytest tests/ -v to execute all pipeline tests')Проверки как живая документация
Каждая проверка в конвейере одновременно служит защитой и частью документации: в машиночитаемой форме она указывает, каким должны быть данные на этом этапе. Когда новый аналитик присоединяется к проекту и читает код конвейера, проверки сообщают ему о контрактах данных без отдельного документа со спецификацией. Относитесь к каждой проверке так же, как к комментарию: формулируйте сообщение достаточно подробно, чтобы было понятно, какое бизнес-правило оно обеспечивает.
# These assertions document business rules as code:
assert (df['order_date'] >= pd.Timestamp('2020-01-01')).all(), \
'Company was founded 2020-01-01; no orders can predate this'
assert df['region'].isin({'North', 'South', 'East', 'West', 'Central'}).all(), \
'Only 5 sales regions are valid; new regions require config update'
print('Business rules verified as assertions.')Быстрая проверка
Проверьте, насколько хорошо Вы усвоили концепции анализа данных из этого урока.
Итоги урока
В этом уроке Вы узнали, как встраивать проверки количества строк, наличия столбцов, диапазона значений и пустых значений непосредственно в конвейер, писать модульные тесты для отдельных функций преобразования с охватом граничных случаев, а также запускать интеграционные тесты и рассматривать проверки как живую документацию контрактов данных. Далее мы рассмотрим планирование и журналирование запусков конвейера для автоматического ежедневного выполнения.
Часто задаваемые вопросы
Урок «Тестирование этапов конвейера с помощью утверждений» бесплатный?
Да — полный текст урока «Тестирование этапов конвейера с помощью утверждений» бесплатно доступен здесь в веб-версии. Чтобы практиковать его интерактивно (встроенный редактор кода и ИИ-репетитор 24/7) и разблокировать остальной курс Pandas & NumPy Academy, подпишись на CoddyKit PRO. Курс Pandas & NumPy Academy содержит 4 уроков всего.
Чему я научусь в уроке «Тестирование этапов конвейера с помощью утверждений»?
Добавляйте на каждом этапе проверки количества строк, утверждения об отсутствии пустых значений и проверки ожидаемых столбцов, чтобы ошибки выявлялись немедленно. Ты практикуешь Pandas & NumPy Academy с помощью реального кода, который запускаешь прямо в браузере, и ИИ-репетитор 24/7 отвечает на твои вопросы во время урока.
Нужен ли мне опыт, чтобы начать Pandas & NumPy Academy?
Предыдущий опыт не требуется. Pandas & NumPy Academy на CoddyKit структурирован для всех уровней — от новичков до продвинутых, поэтому ты можешь начать отсюда или с самого начала и учиться в своем темпе. Это урок 3 из 4.
Сколько времени занимает урок «Тестирование этапов конвейера с помощью утверждений»?
Большинство уроков CoddyKit занимают около 5–10 минут. Каждый из них компактный и интерактивный, поэтому ты постоянно делаешь прогресс и продолжаешь с того же места в веб-версии и приложении.
Можно ли писать и запускать код в этом уроке Pandas & NumPy Academy?
Да. Каждый урок Pandas & NumPy Academy включает встроенный редактор кода, поэтому ты пишешь и запускаешь реальный код прямо в браузере и получаешь моментальную обратную связь от AI — локальная установка не требуется.
Все уроки этого курса
- Оформление шагов преобразования в виде функций
- Параметризация конвейеров с помощью словарей конфигурации
- Тестирование этапов конвейера с помощью утверждений
- Планирование и журналирование запусков конвейера