Тестирование конвейеров данных

1) Зачем тестировать конвейеры данных

• Достоверность (correctness) и стабильность (resilience).
• Предсказуемость изменений (schema/logic evolution).
• Соблюдение SLO по свежести, полноте, латентности.
• Быстрый выпуск (скорость релизов) за счет автоматизированной проверки.

2) Пирамида тестирования данных

Снизу вверх: много быстрых локальных тестов → меньше интеграционных → немного end-to-end.

1. Юнит-тесты трансформаций (функции, UDF, SQL-ви́ды, dbt-модели).
2. Тесты качества данных (правила свежести/полноты/уникальности/диапазонов).
3. Контракты и схемы (schema/contract tests, эволюция).
4. Интеграционные тесты пайплайна (DAG: ingest ↔ storage ↔ transform ↔ marts).
5. E2E-тесты (от источника до витрины/API), включая права (RLS/CLS) и экспорт.
6. Нагрузочные/производительность (объем, скорость, cost-to-serve).
7. Хаос-тесты данных (задержки, дубликаты, out-of-order, недоступность).

3) Типы тестов: что именно проверяем

3.1 Юнит-тесты логики

Чистые функции трансформаций; property-based (инварианты: идемпотентность, монотонность).
SQL/DBT: сравнение результата с эталоном (golden set), запрет `SELECT `, проверка фильтра по времени.

3.2 Тесты качества данных (DQ)

Свежесть: задержка витрин ≤ целевого порога.
Полнота: ожидаемое количество/доля заполненности.
Уникальность: ключи без дубликатов.
Доменные правила: диапазоны, референциальная целостность, бизнес-инварианты.
Аномалии: outliers, всплески дубликатов, разрывы времени.

3.3 Контракты и схемы

Совместимость изменений (SemVer: MAJOR/MINOR/PATCH).
Наличие обязательных колонок, типов, ограничений.
Зафиксированные семантики KPI: формулы и окна агрегаций.

3.4 Интеграционные и E2E

Целостность DAG: триггеры, зависимости, идемпотентный повтор.
Полный путь: источник → raw → curated → marts → BI/API; RLS/CLS.

3.5 Производительность и затраты

p95/p99 латентности джобов, throughput (rows/s), объем/стоимость.
Тесты регрессий производительности и лимиты на сканы.

3.6 Безопасность и приватность

Маскирование PII/PCI (детерминированная токенизация).
Проверка RLS/CLS: пользователи видят только свое.
Экспорт/снепшоты: отсутствие “сырых” персональных полей.

4) Специфика стриминга (Kafka/Flink/Spark Structured Streaming)

Watermarks и lateness: тесты окон с опоздавшими событиями (T+Δ), корректные перерасчеты.
Exactly-once по смыслу: дедуп по `event_id`, идемпотентная запись (upsert/merge).
Out-of-order: инварианты на агрегаты по `event_time`; фиксируем `ingested_at`.
Потеря/повтор: симулируем дроп/реигру партиций, проверяем корректность витрин.

5) Идемпотентность и детерминизм (что и как тестировать)

Повторный запуск шага дает тот же результат (при одинаковых параметрах окна).
Запись — через staging и atomic swap.
Merge-логика с SCD1/SCD2 покрыта тестами конфликтов (last-write-wins, source priority).
Детерминированность UDF/агрегатов: одинаковые входы → одинаковые выходы.

6) Управление тест-данными

Golden datasets: малые эталоны с ручной валидацией.
Синтетика + фабрики данных: покрытие краев домена (nulls, extreme values, Unicode, TZ).
Де-идентифицированные прод-сэмплы: соответствие приватности.
Слоистые фикстуры: сырые события, промежуточные слои, итоговые витрины.

7) Контракты данных: пример и правила

yaml dataset: orders schema:
- name: order_id; type: string; unique: true
- name: user_id; type: string; not_null: true
- name: amount; type: decimal(18,2); min: 0
- name: event_time; type: timestamp; tz: UTC freshness_sla: 10m dq_rules:
- "pct_null(user_id) < 0. 1%"
- "duplicates(order_id) = 0"
- "sum(amount) >= 0"
evolution:
allowed_minor_additions: true breaking_changes_require: approval: 'data-governance'
actions_on_violation:
- quarantine_partition
- replay_last_60m

8) Наблюдаемость и тесты SLO

Экспорт метрик: Freshness, Completeness, Uniqueness, Latency в Grafana/Prometheus.
SLO-алерты как “красные” юнит-тесты в проде (Synthetics).
Репорты регрессий: “после релиза X p95 ↑ на 40%”.

9) CI/CD и среды

CI: юнит+DQ+контракты на PR; schema-diff; статический анализ SQL (линтер).
Песочница / staging: прогон интеграционных и e2e, хаос-тестов с безопасными данными.
Feature-flags: канареечные джобы/модели/формулы.
Каталогизация: версия схем, формул KPI, lineage; автоматическое обновление документации.

10) Хаос-тестирование данных (Chaos-Data)

Инъекция дубликатов/пропусков, задержек, out-of-order.
Падение брокера/партиции, “битые” файлы, schema drift.
Валидируем: авто-ремонт (replay/backfill), quarantine и алерты, MTTR-data.

11) Нагрузочное и стоимость

Генераторы трафика с профилем p95/пики.
Лимиты на скан/шаг (bytes scanned, time caps).
A/B профилировщик стоимости: “старая” vs “новая” модель/запрос.

12) Инструменты (примерные классы)

DQ/Контракты: dbt tests, Great Expectations, Deequ, Soda, Custom linters.
Оркестрация: Airflow/Dagster/Argo/Prefect (операторы для тестов на каждом шаге).
Платформы: BigQuery/Snowflake/Redshift/ClickHouse/Delta/Iceberg/Hudi.
Стриминг: Kafka, Flink, Spark Streaming; TestContainers для локальных сред.
Observability: Prometheus/Grafana/Otel; каталоги DataHub/Amundsen/Collibra.

13) Антипаттерны

“Тестировать нечего — это просто SQL”: нет юнитов и DQ → ломаются метрики.
Только E2E: медленно, нестабильно, причины поломок не ясны.
SELECT: ломается при MINOR-эволюции.
Live-чтение OLTP в тестах: нестабильность и флейки.
Отсутствие golden-наборов: нечем сравнивать результаты.
Нет тестов идемпотентности: повторный запуск портит данные.
Забытый стриминг: не тестируются lateness/out-of-order/повторная доставка.

14) Дорожная карта внедрения

1. Базис: юнит-тесты трансформаций, golden-наборы, линтер SQL, DQ-правила top-10 витрин.
2. Контракты: schema-diff в CI, SemVer, автоматические проверки совместимости.
3. Интеграции: DAG-тесты, idempotency, e2e для критических потоков.
4. Стриминг: тесты watermarks/lateness, dedup/idempotent sinks.
5. SLO и хаос: метрики качества в проде, алерты, хаос-сценарии, MTTR-цели.
6. Оптимизация: перф-регрессии, бюджет-гварды, канареечные релизы.

15) Чек-лист перед релизом

• Юнит-тесты покрывают ключевые трансформации и UDF.
• DQ-правила для свежести/полноты/уникальности/диапазонов проходят.
• Контракты и schema-diff зеленые; нет ломаюших изменений без аппрува.
• Идемпотентность проверена; atomic sink/merge работает.
• Стриминг: watermarks/late data/out-of-order покрыты; dedup на месте.
• SLO-метрики экспонируются; алерты сконфигурированы; runbooks есть.
• Тест-данные безопасны; PII маскированы; RLS/CLS проверены.
• Перф-регрессий нет; лимиты на сканы/время соблюдены.
• Хаос-тесты базовых сценариев прошли; MTTR-целевой достижим.

16) Примеры мини-шаблонов

16.1 Юнит-тест SQL (псевдо-dbt):

sql
-- tests/assert_positive_amount. sql select count() as c from {{ ref('fct_payments') }}
where amount < 0 having c = 0

16.2 Правило свежести (Great Expectations-стиль):

yaml expect_table_row_count_to_be_between:
min_value: 1000 mostly: 0. 99 expect_column_values_to_not_be_null:
column: user_id expect_column_values_to_be_unique:
column: txn_id

16.3 Проверка lateness в стриме (псевдо-код):

python emit(events_out_of_window <= threshold)
emit(reprocessed_events == late_events_detected)

16.4 Contract-test (schema-diff CI):

bash schema-diff --current models/orders. yml --target prod_schema. yml --semver

17) Итог

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