¶ Введение
ETL (Extract, Transform, Load) — фундаментальный процесс в управлении данными.
Конвейер, по которому данные перемещаются из различных источников в единое хранилище для анализа
Extract (Извлечение)
Данные забираются из исходных систем.
Базы данных (SQL), CRM, ERP-системы (1C, SAP), плоские файлы (CSV, Excel), API, логи приложений
Задача: считывать данные, минимально нагружая рабочие системы
Transform (Преобразование)
В одной системе дата записана как "2026/01/01", а в другой — "1 янв 2026").
Очистка: удаление дубликатов, заполнение пропусков
Стандартизация: приведение валют, единиц измерения и форматов дат к единому виду
Агрегация: суммирование продаж по дням или регионам
Обогащение: добавление данных из справочников (по названию города подтягиваются широта и долгота)
Load (Загрузка)
Обработанные данные загружаются в целевую систему.
Куда: хранилище данных (Data Warehouse / DWH), Greenplum, ClickHouse или PostgreSQL
Сегодня также набирает популярность подход ELT (Extract, Load, Transform). Данные сначала загружаются в хранилище "как есть", а преобразование происходит уже внутри хранилища мощностями самого облака. Если потребуется изменить логику расчетов, то не нужно заново выгружать данные из источников — они уже лежат в "сыром" слое хранилища.
Надежность конвейеров Extract, Transform, Load (ETL) не может оцениваться исключительно по успешности выполнения программного кода. Успешное завершение процесса с кодом возврата 0 не гарантирует семантическую корректность данных.
Традиционная парадигма тестирования ПО, ориентированная на проверку отсутствия исключений во время выполнения, кажется несостоятельной при работе с данными. Конвейер может успешно отработать, но на обновленном дашборде мы увидим искаженные финансовые показатели или смещенные выборки для обучения моделей машинного обучения.
¶ Структурные уязвимости процессов ETL и ELT
Структура конвейеров данных содержит специфические классы уязвимостей. В классическом процессе ETL данные извлекаются из разнородных источников (API, SQL-базы, плоские файлы), трансформируются к единой схеме и загружаются в целевое хранилище. Каждый из этих этапов несет свои риски, требующие специализированных подходов к тестированию.
На этапе извлечения (Extraction) уязвимости часто носят внешний характер. Одной из наиболее часто встречающихся проблем является дрейф схемы (Schema Drift), когда система-источник без предупреждения изменяет тип данных столбца или удаляет поле.
Это может привести к тому, что числовые поля будут интерпретироваться как строковые, или возникнет массовое заполнение значений NULL, что нарушит работу аналитических систем. Также ограничения API или скрытые сетевые таймауты могут привести к частичному (а не полному) извлечению данных. Такие сбои часто вызывают аномалии объема, с технической точки зрения запрос к API был выполнен успешно, но вернул неполный набор записей.
Этап преобразования (Transformation) является местом возникновения наиболее острых ошибок. Это логические ошибки в оконных функциях SQL, потеря точности при агрегации чисел с плавающей точкой и некорректная обработка временных зон. Создаются данные, которые выглядят правильными по структуре, однако фактически они являются ложными. Ошибки опасны тем, что они накапливаются со временем, их обнаружение зачастую происходит спустя месяцы.
На этапе загрузки (Load) бывает возникают нарушения ограничений уникальности или внешних ключей. В хранилищах / озерах данных соответствующие ограничения зачастую не проверяются на уровне системы хранения в целях повышения производительности записи. Это перекладывает ответственность за валидацию на логику конвейера, требуя явных проверок перед отдачей данных в аналитическую среду.
¶ Ошибки трансформации
¶ Оконные функции в SQL
Одной из самых распространенных ошибок в трансформациях временных рядов и скользящих агрегациях является неправильное понимание механизмов фрейминга в оконных функциях SQL. При расчете кумулятивных сумм или скользящих средних инженеры часто полагаются на поведение по умолчанию, что приводит к некорректным результатам при наличии дублирующихся временных меток.
Стандарт
SQLопределяет, что если в оконной функции присутствуетORDER BY, но не указан фрейм, то по умолчанию используется:RANGE
Различие в том, как режимы RANGE и ROWS обрабатывают строки с одинаковыми значениями в столбце сортировки. Режим ROWS оперирует физическими строками, где CURRENT ROW означает конкретную текущую строку обработки. Режим RANGE оперирует логическими значениями, в этом случае CURRENT ROW включает все строки, имеющие то же значение сортировки, что и текущая строка.
Пусть конвейер вычисляет нарастающий итог транзакций. Если две транзакции произошли в одну и ту же миллисекунду, то фрейм RANGE включит обе транзакции в сумму для первой из этих строк. Это создает эффект скачкообразного возрастания суммы до обработки второй строки, что нарушает логику последовательного накопления.
| Временная метка | Значение | Кумулятивная сумма (ROWS — ОК) | Кумулятивная сумма (RANGE — Ошибка) |
|---|---|---|---|
| 08:00:00 | 10 | 10 | 30 (10 + 20) |
| 08:00:00 | 20 | 30 (10 + 20) | 30 (10 + 20) |
| 08:01:00 | 30 | 60 | 60 |
¶ Потеря точности и накопление ошибок в числах с плавающей точкой
Использование чисел с плавающей точкой (стандарт IEEE 754) для финансовых агрегаций приводит к критическим проблемам. В большинстве языков программирования (Python,C#) и SQL-движков тип float или double представляют приближенные значения. Классический пример: 0.1 + 0.2 дает 0.30000000000000004 вместо 0.3. Последствия этого в масштабах Big Data часто недооцениваются.
Единичная ошибка может показаться нестрашной, но в процессе ETL (когда суммируются миллионы строк) происходит накопление расхождений. Проблема заключается в нарушении ассоциативности сложения для чисел с плавающей точкой. Математическое тождество не выполняется в компьютерной арифметике, если порядки величин существенно различаются. Это означает, что результат суммирования большого массива чисел может изменяться просто от изменения порядка сортировки входных данных (при распределенной обработке в Spark), что делает результат недетерминированным.
Пусть у float есть ограничение точности (может запомнить только 7 значащих цифр). Попробуем сложить числа: , , .
Сценарий 1: Сначала большие
Компьютер пытается прибавить 0.00005 к миллиону. Но у него не хватает ячеек, чтобы записать такую мелочь на фоне миллиона. Происходит потеря значимости. Итог: . (потеряли небольшие числа).
Другой сценарий:
Сначала складываем небольшие числа. Получаем . Теперь прибавляем это к миллиону. Число достаточно велико, чтобы зацепиться за край точности и учитываться. По итогу: .
Другой формой ошибки является катастрофическая отмена (Catastrophic Cancellation), которая возникает при вычитании двух почти равных чисел и приводит к потере значимых разрядов при расчете дисперсии или ковариации. Это особенно актуально для алгоритмов, вычисляющих дельты между версиями данных.
Истинное в памяти 10.0000
Истинное в памяти 10.0000
Операция: ну вы поняли
До начала агрегации необходимо убедиться, что для финансовых полей используются типы данных
DECIMALилиNUMERICс фиксированной точностью
Требуется отказ от проверок на строгое равенство в пользу проверок с порогом допуска
¶ Конвертация временных зон
Глобализация данных создает риски. Распространенным сценарием является неявное преобразование наивных временных меток (без информации о зоне) в осведомленные (с зоной). Если система-источник экспортирует данные в EST, а слой трансформации интерпретирует их как UTC, то возникает смещение данных во времени.
Более сложным случаем являются граничные условия переходов на летнее/зимнее время (DST). Когда часы переводятся назад, временной интервал дублируется (1:30 ночи наступает дважды), что может привести к дублированию записей в конвейере. При переводе часов вперед образуется разрыв, что может быть ложно интерпретировано мониторингом как потеря данных.
Есть магазин в Нью-Йорке (часовой пояс EST, UTC-5). 1 января в 10:00 утра происходит продажа. База данных магазина сохраняет это как наивную метку времени (без указания пояса): 2025-01-01 10:00:00.
ETL-процесс забирает эти данные в единое хранилище (DWH), которое работает в UTC. ETL видит 10:00:00 и просто приписывает этому времени ярлык UTC. В результате транзакция переместилась на 5 часов в прошлое.
Рассмотрим другую ситуацию.
Осень, когда заканчивается летнее время (DST). Часы переводятся с 03:00 на 02:00. Интервал с 02:00 до 03:00 проживается дважды.
Есть IoT-датчик, который отправляет данные каждые 30 минут. Он шлет наивное локальное время.
Поток данных:
- 02:30 -> timestamp='02:30', value=10
- часы переводятся
- 02:30 -> timestamp='02:30', value=15
Если в базе данных стоит ограничение уникальности PRIMARY KEY (sensor_id, timestamp), то загрузка второй записи упадет с ошибкой. Если ограничений нет, и делаем SUM(value), то посчитаем данные за этот час дважды.
Еще один сценарий.
Весной часы переводятся с 02:00 на 03:00. Времени 02:30 в этот день просто не существует.
Настроен мониторинг, который проверяет целостность данных. Если в течение часа не поступило ни одной записи о продажах — тревога.
Поток данных:
- 01:59 -> запись есть
- часы переводятся на 03:00
- 03:01 -> запись есть
Не следует отбрасывать информацию о часовом поясе на этапе Extract
При этом:
- на входе если источник дает 2025-01-01 10:00, мы должны знать пояс и сразу преобразовать в
timestamp with time zone - в хранилище хранить все строго в UTC
- на витрине конвертировать из UTC в локальное время пользователя только в самый последний момент, при отображении отчета
¶ Основы профилирования данных
Помимо структурных ошибок конвейер может потерять свою ценность из-за изменения характера данных. Явление, известное как дрейф данных (data drift) или дрейф концепции (concept drift). Задача тестирования в данном случае — определить, насколько распределение нового пакета данных отличается от эталонного.
Дивергенция Кульбака-Лейблера (относительная энтропия) представляет собой меру различия между двумя вероятностными распределениями: (истинное или эталонное распределение) и (наблюдаемое или новое распределение).
Для дискретных вероятностных распределений и , определенных на одном и том же вероятностном пространстве , дивергенция KL определяется формулой:
Значение показывает информационные потери при попытке аппроксимировать эталон с помощью новых данных . Если распределения идентичны, отношение под логарифмом равно 1, логарифм равен 0 и дивергенция равна нулю.
Существует математическая тонкость, критичная для ETL: если для некоторого события вероятность в эталоне , а в новом батче (или стремится к нулю), то дивергенция стремится к бесконечности. Событие, считавшееся возможным в прошлом, теперь стало невозможным. В практических реализациях для избежания деления на ноль применяется сглаживание — добавление малого значения ко всем весам.
Дивергенция KL является теоретическим фундаментом, но также часто используется производный от нее Индекс стабильности популяции (Population Stability Index, PSI)
PSI является симметричной метрикой, которая суммирует взвешенные разности распределений.
Алгоритм расчета PSI для автоматизированного тестирования включает следующие шаги:
- Сортировка переменной скоринга или анализируемого атрибута.
- Разбиение данных на бины на основе эталонной выборки.
- Расчет процента записей в каждом бине для эталонной () и текущей () выборок.
- Применение формулы:
Здесь:
- — количество бинов, на которые разбиты данные
- — доля (%) наблюдений в -м бине в текущем наборе данных
- — доля (%) наблюдений в -м бине в эталонном наборе данных
Значения и используются в долях, а не в процентах (не 20%, а 0.2).
Для автоматизации решений в конвейерах (Pass/Fail) используются эвристические пороговые значения:
- PSI < 0.1: изменения незначительны, конвейер продолжает работу
- 0.1 PSI < 0.2: умеренный дрейф, генерируется предупреждение
- PSI 0.2: значительный дрейф, конвейер должен быть остановлен, так как данные больше не репрезентативны
Пример.
Есть скоринговая модель, которая оценивает надежность клиента от 0 до 1000 баллов.
Эталон (Expected): данные за прошлый год, на которых модель училась.
Текущие данные (Actual): заявки, пришедшие за последнюю неделю после запуска новой рекламной кампании.
Нужно понять, не стала ли новая аудитория слишком рискованной (дрейф данных), что может привести к массовым невозвратам кредитов.
| Бин | Описание | Доля в Эталоне () | Доля в Текущем () | Что произошло |
|---|---|---|---|---|
| 1 | Высокий риск | 0.20 (20%) | 0.35 (35%) | Пришло много рискованных клиентов |
| 2 | Средний риск | 0.50 (50%) | 0.45 (45%) | Немного меньше обычного |
| 3 | Надежные | 0.30 (30%) | 0.20 (20%) | Надежных стало меньше |
Бин 1:
- разница:
- отношение:
- логарифм:
- итог:
Бин 2:
- разница:
- отношение:
- логарифм:
- итог:
Бин 3 (Надежные):
- разница:
- отношение:
- логарифм:
- итог:
¶ Фреймворк Great Expectations
Ведущий Python-фреймворк для валидации, документирования и профилирования данных. Архитектура GX строится вокруг иерархической объектной модели.
Элементы в соответствие с иерархической моделью.
Data Context (Контекст данных)
Точка входа. Контекст управляет конфигурацией всех источников данных, хранилищ ожиданий и результатов валидации.
Data Source & Data Asset (Источник и Актив)
Определяют, где находятся данные (Pandas DataFrame, таблица SQL, Spark DataFrame) и как к ним подключиться.
Batch Definition (Определение батча)
Формализует способ извлечения среза данных (весь датафрейм или часть за конкретный год). Один и тот же BatchDefinition всегда будет обращаться к данным одним и тем же способом (воспроизводимость тестов)
Expectation Suite (Набор ожиданий)
Коллекция конкретных утверждений (Expectations), применяемых к данным.
Validation Definition (Определение валидации)
Связывает BatchDefinition (что тестировать) с ExpectationSuite (как тестировать).
Checkpoint (Контрольная точка)
Исполнительный механизм. Запускает одно или несколько Определений Валидации, выполняет действия по результатам (обновление документации или отправка уведомлений) и возвращает результаты.
Пример для Google Colab:
import great_expectations as gx
import pandas as pd
import shutil
import os
from google.colab import files
context = gx.get_context()
# создание образца данных
# Исправлено: добавлены значения для transaction_id и currency
df = pd.DataFrame({
"transaction_id": [101, 102, 103, 104, 105],
"amount": [100.50, 200.00, 150.75, 400.20, 50.00],
"currency": ["USD", "USD", "USD", "RUB", "USD"], # 'RUB' вызовет ошибку валидации
"timestamp": pd.to_datetime(["2025-01-01", "2025-01-02", "2025-01-03", "2025-01-04", "2025-01-05"])
})
# источник и актив
data_source = context.data_sources.add_pandas("my_pandas_source")
data_asset = data_source.add_dataframe_asset(name="transactions_asset")
# Определение батча (весь фрейм)
batch_definition = data_asset.add_batch_definition_whole_dataframe("full_dataframe_batch")
# Те самые Великие ожидания...
suite = context.suites.add(gx.ExpectationSuite(name="transaction_suite"))
suite.add_expectation(
gx.expectations.ExpectColumnValuesToBeInSet(
column="currency",
value_set=["USD"]
)
)
suite.add_expectation(
gx.expectations.ExpectColumnValuesToNotBeNull(column="transaction_id")
)
suite.add_expectation(
gx.expectations.ExpectColumnValuesToBeBetween(
column="amount",
min_value=0,
max_value=1000
)
)
validation_definition = context.validation_definitions.add(
gx.ValidationDefinition(
name="transaction_validation",
data=batch_definition,
suite=suite
)
)
checkpoint = context.checkpoints.add(
gx.Checkpoint(
name="daily_checkpoint",
validation_definitions=[validation_definition],
actions=[
gx.checkpoint.actions.UpdateDataDocsAction(name="update_data_docs")
]
)
)
validation_results = checkpoint.run(batch_parameters={"dataframe": df})
print(f"Success: {validation_results.success}")
# экспорт Data Docs
sites = context.get_docs_sites_urls()
full_path_to_html = sites[0]["site_url"].replace("file://", "")
if full_path_to_html.endswith(".html"):
site_dir = os.path.dirname(full_path_to_html)
else:
site_dir = full_path_to_html
output_filename = "gx_data_docs_report"
shutil.make_archive(output_filename, 'zip', site_dir)
files.download(f"{output_filename}.zip")
Одной из самых мощных возможностей Great Expectations является способность генерировать ожидания автоматически (Data Assistants).
GX также предоставляет дистрибутивные ожидания (Distributional Expectations).
Для эффективного внедрения описанных инструментов рекомендуется использовать стратегию многоуровневого тестирования. Не все тесты должны запускаться на каждом этапе
Уровень 1: Критические проверки схемы и типов
Должны выполняться для каждого батча.
Все ожидаемые столбцы существуют и их типы соответствуют контракту. Проверка первичных ключей и критически важных внешних ключей на отсутствие NULL. Проверка уникальности идентификаторов.
Уровень 2: Верификация бизнес-логики и трансформаций
Верифицируют специфическую логику ETL-конвейера.
Использование ExpectColumnValuesToBeInSet для кодов статусов, валют или стран. Валидация того, что сумма транзакций в таблице фактов равна ежедневному итогу в таблице агрегатов. Использование SQL-ожиданий (QueryExpectation). Верификация того, что скользящие суммы ведут себя ожидаемо.
Уровень 3: Статистическое профилирование
Контроль аномалий объема. Если ежедневная загрузка обычно содержит 1 миллион строк, а сегодня пришло 10 тысяч, это явный признак сбоя в источнике или на этапе экстракции. Использование ожиданий KL Divergence или PSI для гарантии того, что распределение данных не изменилось.
¶ Заключение
Тестирование конвейеров данных требует изменения мышления: перехода от тестирования логики кода к тестированию состояния данных
Логические ошибки (неточность чисел с плавающей точкой и проблемы фрейминга оконных функций) могут быть частично выявлены строгим модульным тестированием, но динамическая природа входящих данных диктует необходимость непрерывной валидации.
Интеграция фреймворка Great Expectations дает стандартизированный подход к этой задаче. Путем создания Определений Валидации, охватывающих как детерминированные правила (схема, типы), так и вероятностные ожидания (дивергенция KL), тестировщики могут гарантировать стабильность конвейеров.