Зображення, згенероване штучним інтелектом, демонструє інженера даних за роботою у високотехнологічному середовищі, з кількома екранами, що відображають різні аспекти модульного тестування та інженерії даних
У сфері інженерії даних однією з практик, яку часто не беруть до уваги, є модульне тестування (Unit Testing). Багато хто може подумати, що модульні тести – це просто методологія розробки програмного забезпечення, але це далеко не так. Оскільки ми прагнемо створити надійні, безпомилкові конвеєри даних і моделі даних SQL, цінність модульного тестування в інженерії даних стає все більш очевидною. Сьогодні ми заглибимося в те, як можна застосувати ці зрілі практики розробки програмного забезпечення до інженерії даних.
Важливість модульного тестування
Для тих, хто не знайомий, модульне тестування (Unit Testing) – це практика тестування окремих модулів або компонентів програмного забезпечення, щоб переконатися, що кожен з них працює так, як очікується. Це може бути що завгодно – від функцій у скрипті Python до окремих SQL-запитів. Перевіряючи правильність роботи кожної одиниці, ми значно підвищуємо загальну надійність нашого програмного забезпечення, або, в даному випадку, наших конвеєрів даних і моделей даних.
В контексті інженерії даних, застосування модульного тестування гарантує, що ваші дані, а також ваша бізнес-логіка є точними. Це призводить до отримання високоякісних даних, яким можуть довіряти ваші аналітики даних, науковці та особи, що приймають рішення.
Конвеєри даних модульного тестування
Конвеєри даних часто включають складні послідовності операцій вилучення, перетворення та завантаження даних (ETL). Тому ймовірність помилок дуже велика. Для модульного тестування цих операцій ми розбиваємо конвеєр на окремі компоненти і перевіряємо кожен з них незалежно.
Візьмемо простий конвеєр, який витягує дані з CSV-файлу, трансформує їх, очищаючи від нульових значень, і завантажує в базу даних. Ось приклад на Python з використанням панд:
import pandas as pd
from sqlalchemy import create_engine
# Function to load the CSV
def load_data(file_name):
data = pd.read_csv(file_name)
return data
# Function to clean the data
def clean_data(data):
data = data.dropna()
return data
# Function to save the data to a SQL database
def save_data(data, db_string, table_name):
engine = create_engine(db_string)
data.to_sql(table_name, engine, if_exists='replace')
# Run pipeline
data = load_data('data.csv')
data = clean_data(data)
save_data(data, 'sqlite:///database.db', 'my_table')Для модульного тестування цього конвеєра ми напишемо окремі тести для кожної з функцій за допомогою бібліотеки на кшталт pytest.
У цьому прикладі ми маємо три основні функції: load_data, clean_data та save_data. Напишемо тести для кожної з них. Для load_data і save_data нам потрібно створити тимчасовий CSV-файл і базу даних SQLite, що ми можемо зробити за допомогою функції fixture бібліотеки pytest.
import os
import pandas as pd
import pytest
from sqlalchemy import create_engine, inspect
# Use pytest fixtures to set up a temporary CSV file and SQLite database
@pytest.fixture
def csv_file(tmp_path):
data = pd.DataFrame({
'name': ['John', 'Jane', 'Doe'],
'age': [34, None, 56] # Jane's age is missing
})
file_path = tmp_path / "data.csv"
data.to_csv(file_path, index=False)
return file_path
@pytest.fixture
def sqlite_db(tmp_path):
file_path = tmp_path / "database.db"
return 'sqlite:///' + str(file_path)
def test_load_data(csv_file):
data = load_data(csv_file)
assert 'name' in data.columns
assert 'age' in data.columns
assert len(data) == 3
def test_clean_data(csv_file):
data = load_data(csv_file)
data = clean_data(data)
assert data['age'].isna().sum() == 0
assert len(data) == 2 # Jane's record should be removed
def test_save_data(csv_file, sqlite_db):
data = load_data(csv_file)
data = clean_data(data)
save_data(data, sqlite_db, 'my_table')
# Check the data was saved correctly
engine = create_engine(sqlite_db)
inspector = inspect(engine)
tables = inspector.get_table_names()
assert 'my_table' in tables
loaded_data = pd.read_sql('my_table', engine)
assert len(loaded_data) == 2 # Only John and Doe's records should be presentОсь ще один приклад: Припустимо, у вас є конвеєр, який завантажує дані з CSV-файлу і перетворює їх, перетворюючи стовпець «дата» з рядка на час:
def convert_date(data, date_column):
data[date_column] = pd.to_datetime(data[date_column])
return dataЮніт-тест для цієї функції передасть їй DataFrame з відомим форматом рядка для дат. Потім він перевіряє, чи правильно функція перетворює дати в об’єкти типу datetime і чи правильно вона обробляє неприпустимі формати.
Напишемо модульний тест для наведеного вище сценарію. Цей тест спочатку перевіряє функцію з дійсними датами, стверджуючи, що стовпець ‘date’ у вихідному фреймі даних дійсно має тип datetime d, і що значення відповідають очікуваним. Потім він перевіряє, що функція коректно згенерує помилку ValueError, коли отримує невірну дату.
import pandas as pd
import pytest
def test_convert_date():
# Test with valid dates
test_data = pd.DataFrame({
'date': ['2021-01-01', '2021-01-02']
})
converted_data = convert_date(test_data.copy(), 'date')
assert pd.api.types.is_datetime64_any_dtype(converted_data['date'])
assert converted_data.loc[0, 'date'] == pd.Timestamp('2021-01-01')
assert converted_data.loc[1, 'date'] == pd.Timestamp('2021-01-02')
# Test with an invalid date
test_data = pd.DataFrame({
'date': ['2021-13-01'] # This date is invalid because there's no 13th month
})
with pytest.raises(ValueError):
convert_date(test_data, 'date')Наведемо останній приклад: Припустимо, у вас є конвеєр, який завантажує дані, а потім агрегує їх, обчислюючи загальний обсяг продажів для кожного регіону:
def aggregate_sales(data):
aggregated = data.groupby('region').sales.sum().reset_index()
return aggregatedЮніт-тест для цієї функції передасть їй DataFrame з даними про продажі в різних регіонах. Тест перевірить, чи правильно функція обчислює загальну суму продажів для кожного регіону.
Напишемо юніт-тест для цієї функції. У цьому тесті ми спочатку передаємо функції aggregate_sales DataFrame з відомими даними про продажі для кожного регіону і перевіряємо, чи правильно вона агрегує продажі. Потім ми передаємо їй DataFrame без продажів і перевіряємо, чи правильно вона агрегує їх до 0. Це гарантує, що функція правильно обробляє як типові, так і граничні випадки.
Ось як можна написати юніт-тест за допомогою бібліотеки pytest для функції aggregate_sales:
import pandas as pd
import pytest
def test_aggregate_sales():
# Test data with sales for each region
test_data = pd.DataFrame({
'region': ['North', 'North', 'South', 'South', 'East', 'East', 'West', 'West'],
'sales': [100, 200, 300, 400, 500, 600, 700, 800]
})
aggregated = aggregate_sales(test_data)
assert aggregated.loc[aggregated['region'] == 'North', 'sales'].values[0] == 300
assert aggregated.loc[aggregated['region'] == 'South', 'sales'].values[0] == 700
assert aggregated.loc[aggregated['region'] == 'East', 'sales'].values[0] == 1100
assert aggregated.loc[aggregated['region'] == 'West', 'sales'].values[0] == 1500
# Test with no sales data
test_data = pd.DataFrame({
'region': ['North', 'South', 'East', 'West'],
'sales': [0, 0, 0, 0]
})
aggregated = aggregate_sales(test_data)
assert aggregated.loc[aggregated['region'] == 'North', 'sales'].values[0] == 0
assert aggregated.loc[aggregated['region'] == 'South', 'sales'].values[0] == 0
assert aggregated.loc[aggregated['region'] == 'East', 'sales'].values[0] == 0
assert aggregated.loc[aggregated['region'] == 'West', 'sales'].values[0] == 0Модульне тестування моделей даних SQL
Коли мова йде про моделі даних на основі SQL, модульні тести здебільшого зводяться до тестування ваших SQL-запитів. Наприклад, ви можете перевірити, що запит повертає очікувані результати при певних вхідних даних, або що він правильно обробляє крайні випадки.
Розглянемо SQL-запит, який обчислює середню вартість замовлення:
SELECT AVG(order_value) as avg_order_value FROM orders;Юніт-тест для цього запиту може вставити кілька рядків у таблицю замовлень і перевірити, що запит правильно обчислює середнє значення. Існує кілька інструментів для полегшення модульного тестування SQL, але найпопулярнішим у сфері інженерії даних є dbt (data build tool), який має вбудовану підтримку для тестування даних.
У цьому прикладі ми припустимо, що ви створили проект у dbt і дотримуєтесь рекомендованої структури проекту. У вас є модель, яка обчислює середню вартість замовлення (назвемо її avg_order_value.sql), і ми можемо додати тест для неї.
Спочатку визначимо модель в avg_order_value.sql:
-- This is your model which calculates the average order value
SELECT AVG(order_value) as avg_order_value FROM {{ ref('orders') }}Для тестування ми будемо використовувати тести даних dbt. Це просто SQL-запити, які dbt виконує до ваших даних. Якщо запит повертає будь-які рядки, тест не пройдено. Якщо запит не повертає жодного рядка, тест пройдено.
Ми створимо новий файл в каталозі tests з назвою avg_order_value_test.sql. У цьому файлі ми напишемо запит, який перевіряє, чи знаходиться середнє значення замовлення в очікуваних межах:
-- This is your test
-- Assume you've calculated the average manually for your test data and it is 100
WITH avg_order_value_test AS (
SELECT AVG(order_value) as avg_order_value FROM {{ ref('orders') }}
)
SELECT *
FROM avg_order_value_test
WHERE avg_order_value <> 100 -- The test will fail if the average order value is not 100Не забудьте замінити 100 на очікуване середнє значення замовлення для ваших тестових даних.
Ви можете запустити цей тест за допомогою команди dbt test.
Це простий випадок, але ви можете створити більш складні тестові сценарії. Наприклад, ви можете додати більше даних до таблиці orders і перевірити, чи правильно оновлюється середнє значення, або додати граничні випадки, такі як замовлення зі значенням 0 або NULL, і переконатися, що вони обробляються правильно.
Тестування COUNT: Припустимо, у вас є SQL-запит, який підраховує кількість активних користувачів у таблиці:
SELECT COUNT(*) as active_users FROM users WHERE status = 'active';Юніт-тест для цього запиту вставить деякі тестові дані в таблицю користувачів, переконавшись, що відома кількість цих користувачів має статус «активний». Потім він виконає запит і перевірить, що кількість повернених користувачів відповідає очікуваному значенню.
Використовуючи dbt, тест на основі YAML для прикладу COUNT може виглядати так:
version: 2
models:
- name: users
tests:
- dbt_utils.expression_is_true:
expression: "count(*) = 5"
where: "status = 'active'"Тут ми стверджуємо, що є рівно 5 активних користувачів. Змініть це значення відповідно до ваших налаштувань тестових даних.
Тестування з’єднань: Припустимо, у вас є запит, який об’єднує дві таблиці: orders і order_items, підсумовуючи загальну вартість кожного замовлення:
SELECT o.order_id, SUM(i.item_price * i.quantity) as total_order_cost
FROM orders o
JOIN order_items i ON o.order_id = i.order_id
GROUP BY o.order_id;Щоб виконати модульне тестування цього запиту, спочатку потрібно заповнити таблиці orders і order_items тестовими даними, переконавшись, що деякі замовлення містять кілька товарів. Потім тест перевірить, чи правильно розраховано total_order_cost для кожного замовлення, особливо для замовлень з кількома товарами.
Використовуючи dbt, тест на основі YAML для прикладу JOIN може виглядати так:
version: 2
models:
- name: orders
tests:
- dbt_utils.expression_is_true:
expression: "total_order_cost = 100.00"
where: "order_id = 1"Тут ми стверджуємо, що загальна вартість замовлення для order_id 1 дорівнює 100. Відрегулюйте це відповідно до ваших тестових даних.
Давайте зробимо ще одне твердження. У цьому прикладі ми хочемо перевірити, що наша операція об’єднання і подальша операція агрегування працюють коректно. Для того, щоб створити більш складний тест, давайте розглянемо сценарій, де у нас є кілька замовлень, з декількома позиціями в кожному замовленні, і ми хочемо перевірити розрахунок загальної вартості для декількох замовлень.
Припустимо, що наші тестові дані виглядають наступним чином:
Таблиця замовлень:
[
{
"order_id": 1,
"customer_id": 100
},
{
"order_id": 2,
"customer_id": 101
},
{
"order_id": 3,
"customer_id": 102
}
]Order_Items Table:
[
{
"order_item_id": 1,
"order_id": 1,
"item_price": 10,
"quantity": 5
},
{
"order_item_id": 2,
"order_id": 1,
"item_price": 20,
"quantity": 2
},
{
"order_item_id": 3,
"order_id": 2,
"item_price": 30,
"quantity": 3
},
{
"order_item_id": 4,
"order_id": 2,
"item_price": 40,
"quantity": 1
},
{
"order_item_id": 5,
"order_id": 3,
"item_price": 50,
"quantity": 1
}
]У представленні JSON кожен рядок таблиці стає об’єктом JSON всередині масиву. Назви стовпців стають ключами об’єктів JSON, а відповідні значення комірок – значеннями об’єктів JSON. Дивіться таблицю нижче для представлення цих даних.
У цьому сценарії ми очікуємо, що загальна вартість замовлення 1 становитиме 90 (50 + 40), замовлення 2 – 130 (90 + 40) і замовлення 3 – 50 (50*1).
Наш тест на основі dbt YAML може виглядати так:
version: 2
models:
- name: orders
tests:
- dbt_utils.expression_is_true:
expression: "total_order_cost = 90.00"
where: "order_id = 1"
- dbt_utils.expression_is_true:
expression: "total_order_cost = 130.00"
where: "order_id = 2"
- dbt_utils.expression_is_true:
expression: "total_order_cost = 50.00"
where: "order_id = 3"Тут ми додаємо кілька тверджень, щоб перевірити правильність загальної вартості замовлення для кількох замовлень. Це дозволяє нам перевірити не тільки правильність нашої операції об’єднання, але й нашу здатність правильно розрахувати загальну вартість для кількох замовлень і позицій.
Модульне тестування та безперервна інтеграція
Коли у вас є набір модульних тестів, ви можете інтегрувати їх у конвеєр безперервної інтеграції (CI). Конвеєр CI автоматично збирає і тестує ваш код щоразу, коли зміни вносяться до кодової бази. Таким чином, якщо зміна призведе до помилки у вашому конвеєрі даних або моделі даних, конвеєр CI виявить її до того, як вона вплине на ваші виробничі системи даних.
Декілька інструментів CI можуть допомогти налаштувати це, наприклад, Jenkins, Travis CI або GitLab CI. Особливості налаштування конвеєра CI залежать від інструменту, але загальна ідея полягає в тому, щоб запускати юніт-тести кожного разу, коли виштовхується новий код.
Давайте подивимося, як можна інтегрувати юніт-тест Python в простий CI-пайплайн за допомогою популярного інструменту CI/CD: GitHub Actions.
Припустимо, у вас є Python-проект з набором модульних тестів, написаних за допомогою pytest. У вас є файл test_example.py з такою функцією:
def test_addition():
assert 1 + 1 == 2Ви можете запускати ці тести автоматично щоразу, коли вносите зміни до вашого репозиторію GitHub, створивши робочий процес GitHub Actions. Створіть новий файл у вашому сховищі за адресою .github/workflows/ci.yml з наступним вмістом:
name: Python CI
on:
push:
branches: [ main ]
pull_request:
branches: [ main ]
jobs:
build:
runs-on: ubuntu-latest
steps:
- uses: actions/checkout@v2
- name: Set up Python
uses: actions/setup-python@v2
with:
python-version: '3.8'
- name: Install dependencies
run: |
python -m pip install --upgrade pip
pip install pytest
# If your project has other dependencies listed in a requirements.txt, install them with:
# pip install -r requirements.txt
- name: Run tests
run: |
pytestЦей робочий процес запускається щоразу, коли ви вносите зміни до основної гілки або відкриваєте pull-запит до main. Він налаштовує свіже середовище Python 3.8, встановлює залежності вашого проекту і запускає ваш набір тестів pytest. Якщо будь-який з ваших тестів не пройде, робочий процес завершиться невдачею, і ви отримаєте повідомлення про це.
Пам’ятайте, що це лише простий приклад. Реальні конвеєри CI будуть складнішими і включатимуть такі речі, як створення образів Docker, розгортання вашого коду в середовищах, запуск наскрізних тестів тощо.
Висновок
Юніт-тестування може здатися додатковою роботою, але попередні зусилля приносять дивіденди в довгостроковій перспективі. У сфері, де довіра до даних має першорядне значення, впевненість у ваших конвеєрах даних і моделях даних є надзвичайно важливою.
Основний висновок полягає в тому, що інженерія даних може отримати значну користь від методологій програмної інженерії. Впроваджуючи такі практики, як модульне тестування та безперервна інтеграція, ми можемо підвищити надійність наших систем даних, виявити помилки до того, як вони поширяться далі, і, зрештою, сприяти розвитку культури довіри до даних в наших організаціях.
Пам’ятайте, що метою цих тестів є не лише перевірка коректності за звичайних умов, але й забезпечення того, щоб ваш конвеєр або запит безболісно обробляв граничні випадки та помилки. Завдяки модульному тестуванню ви можете створювати конвеєри даних і моделі даних SQL, які є надійними і надійними, підвищуючи довіру до ваших систем даних.
Вдалого тестування!
ОРИГІНАЛ СТАТТІ:Unit Testing in Data Engineering: A Practical Guide
АВТОР СТАТІ:Sam Zamany
🚀Долучайтесь до нашої спільноти Telegram:
🚀Долучайтесь до нашої спільноти FaceBook:
