Data Engineering Design Patterns, глава 9: Паттерны качества данных

Главная мысль: Качество данных строится в три слоя. Сначала не пускаем плохие данные дальше (enforcement). Затем держим схему под контролем. Затем наблюдаем за данными, потому что правила сегодня могут не подойти данным завтра.

Доверие к данным легко потерять. Одна неверная цифра в отчёте, и аналитики перестают верить таблице. Глава даёт шесть паттернов в трёх категориях.

Все паттерны главы

Паттерн	Когда применять
`Audit-Write-Audit-Publish` (AWAP)	Нужно остановить пайплайн, если входные или выходные данные плохие
`Constraints Enforcer`	Хочется переложить проверки на базу или формат файла
`Schema Compatibility Enforcer`	Продюсер может сломать схему и уронить потребителей
`Schema Migrator`	Нужно переименовать поле или сменить тип, не ломая потребителей
`Offline Observer`	Нужно следить за данными, но не блокировать основной пайплайн
`Online Observer`	Нужно узнать о проблеме сразу после записи данных

Quality Enforcement: не пустить плохие данные

Первая категория ставит контроль прямо в поток данных или в хранилище.

Audit-Write-Audit-Publish (AWAP)

Проблема. Ежедневный job посчитал уникальных посетителей на 50% меньше. Продукт решил, что это реальное падение, и запустил маркетинговую кампанию. Ошибка была в агрегации. Нужна защита от такого.

Как работает:

Это аналог assert в unit tests, но на реальных данных.
Первый аудит идёт до трансформации. Он быстрый и дешёвый: формат файла, размер, схема, первая строка.
Второй аудит идёт после трансформации и до публикации. Он смотрит на сами данные: NULL, доли, объём, уникальность.
Проверки бывают на уровне записи и на уровне датасета (объём, доля NULL).
Не дублируйте одну проверку в обоих аудитах. NULL лучше проверить один раз, во втором. Тогда он поймает проблему и из входа, и из вашей логики.

Провал аудита не всегда значит остановку:

Реакция	Что делаем
Fail	Останавливаем пайплайн
Data dispatching	Валидные записи идут дальше, невалидные уходят в отдельное хранилище
Nonblocking audit	Публикуем данные с пометкой о проблеме. Потребитель сам решает, использовать ли их

Data dispatching похож на Dead-Letter. Разница: здесь нет неожиданной ошибки. Мы сами решили отделить плохие строки по правилу.

Для streaming есть два варианта:

Window-based: job закрывает окно и запускает аудит над буфером окна.
Staging-based: job пишет результат в staging-слой. Отдельный audit job проверяет его и переносит в финальное место.
Первого аудита обычно нет. Данные идут непрерывно, проще проверять после трансформации.

Минусы:

Лишние затраты на вычисления. Проверки по метаданным дешёвые, построчные дорогие.
Правила покрывают не всё. Данные меняются, и правило может устареть.
В streaming растёт latency. Нужно дождаться конца окна.
Срабатывание аудита не всегда реальная проблема. Резкий рост трафика после вирусного поста тоже выглядит как аномалия. Иногда достаточно alert.

Короткий пример на Python (данные придуманы):

def audit_input(path, min_size=1_000):
    errors = []
    if os.path.getsize(path) < min_size:
        errors.append("file is too small")
    if errors:
        raise Exception("Audit failed: " + "; ".join(errors))

В Airflow это отдельные задачи: audit_input >> transform >> audit_output >> load.

Constraints Enforcer

Проблема. Пайплайн работает месяцами, потом в обязательных полях появляются NULL. Логика job уже сложная. Не хочется добавлять в неё ещё валидацию.

Как работает:

Выберите атрибуты, которым нужны ограничения. Это зависит от бизнеса и закона.
Назначьте ограничения базе или формату файла.
Загрузка падает, если данные их нарушают.

Четыре типа ограничений:

Тип	Что проверяет	Пример
Type	Тип значения	`visit_id` всегда строка
Nullability	Поле не может быть пустым	`NOT NULL`
Value	Значение удовлетворяет выражению	`event_time <= NOW()`
Integrity	Ссылка есть в другой таблице	`page_id` есть в `pages`

Где это есть:

Реляционные базы: NOT NULL, CHECK, foreign key.
Delta Lake: CHECK constraint и NOT NULL.
Avro и Protobuf: типы из коробки, значения через расширения вроде protovalidate.

CREATE TABLE visits (
  visit_id STRING NOT NULL,
  event_time TIMESTAMP NOT NULL
) USING delta;

ALTER TABLE visits ADD CONSTRAINT not_future
  CHECK (event_time < NOW() + INTERVAL 1 SECOND);

Минусы:

All-or-nothing. Одна плохая строка отклоняет всю загрузку.
База часто останавливается на первой ошибке. Чтобы найти все, нужно много попыток.
Ограничения заданы для продюсера. Потребителю могут быть нужны другие правила, и фильтры придётся делать у себя.
Не все правила покрываются. Табличные форматы часто не умеют integrity. Придётся дополнить проверки в коде.

Schema Consistency: схема не должна сломаться

Данные могут быть в порядке много месяцев. Потом продюсер меняет схему, и пайплайн падает. Два паттерна решают это с двух сторон.

Schema Compatibility Enforcer

Проблема. Job сессионизации падал много раз за месяц. Новая команда убрала поля, которые считала ненужными. Нужно запретить ломающие изменения.

Три режима контроля:

Режим	Как работает	Пример
Внешний сервис или библиотека	Продюсер проверяет схему через API до записи	Kafka Schema Registry, Avro `SchemaValidator`
Implicit with inserts	Таблица отклоняет запись, которая не подходит под схему	Delta Lake, реляционные базы
Event-driven DDL	Event trigger проверяет `DROP COLUMN` и `RENAME COLUMN` и откатывает их	PostgreSQL, SQL Server

Если тонкий контроль не нужен, можно просто не выдавать право ALTER TABLE.

Режимы совместимости:

Режим	Что гарантирует	Разрешено
Backward	Новый потребитель читает старые данные	Удалить поле, добавить опциональное
Forward	Старый потребитель читает новые данные	Добавить поле, удалить опциональное
Full	Оба направления	Добавить или удалить опциональное поле

Каждый режим бывает transitive и nontransitive:

Nontransitive проверяет только соседние версии (v2 с v1).
Transitive проверяет все версии (v2 с v1 и v0).
Пример: v0 имеет order_id. v1 добавляет amount с default. v2 делает amount обязательным. Для соседних версий всё в порядке. Но потребитель на v2 не прочитает данные v0, у которых нет amount. В transitive режиме это изменение будет отклонено.

Минусы:

Лишние вызовы. Продюсер проверяет каждую запись по последней схеме.
Эволюция схемы становится сложнее. Любое изменение должно проходить режим. Чтобы переименовать поле, придётся добавить новое и объявить старое устаревшим.

Schema Migrator

Проблема. В событии уже 60 полей, и они разбросаны. Потребители хотят сгруппировать поля пользователя в один объект user. Резкая смена сломает совместимость. Нужен плавный путь.

Как работает:

Определите тип изменения: rename, смена типа или удаление.
Для rename и смены типа сначала добавьте новое поле рядом со старым.
Договоритесь с потребителями о сроке перехода (grace period). Всё это время они получают оба поля.
После срока выпустите схему только с новым полем.
Удаление идёт так же: согласуйте срок, затем выпустите схему без поля.
Режим совместимости не должен быть transitive. Иначе удалять и переименовывать будет нельзя.
Чтобы узнать, кто использует поле, помогает lineage (глава 10, Fine-Grained Tracker).

-- Плохо: потребители упадут сразу
ALTER TABLE visits RENAME COLUMN from_page TO referral;

-- Хорошо: сначала добавляем новое поле
ALTER TABLE visits ADD COLUMN referral VARCHAR(25);
-- старое поле удаляем после срока перехода

Минусы:

Больше данных на время перехода. Растут хранилище, сеть и I/O.
Много полей вредно. Protobuf не советует сотни полей, а Delta Lake по умолчанию собирает статистику только по первым 32 колонкам.
Удаление не всегда возможно. Если поле нужно потребителю и замены нет, его нельзя убрать.

Quality Observation: наблюдать за данными

Правила enforcement покрывают только то, что вы знали заранее. Наблюдение помогает найти новые проблемы и обновить правила. Главное отличие от аудита: аудит блокирует пайплайн, наблюдение не блокирует.

Offline Observer

Проблема. Пайплайн новый и проблем мало. Но вы знаете, что upstream изменится. Хочется следить за распределением значений и числом NULL по колонкам, не мешая основному пайплайну.

Как работает:

Отдельный observability job читает обработанные данные.
Он считает метрики: распределение значений, NULL в nullable-полях, новые поля.
Он пишет метрики в слой мониторинга.
Job идёт по своему расписанию, например ночью, чтобы не конкурировать за ресурсы.
Для идемпотентности job сохраняет состояние: id первой и последней обработанной строки.
В streaming это отдельный job над теми же данными. Он считает те же метрики и может строить отчёт профиля (например, ydata-profiling). Он также считает lag: последний доступный offset минус последний обработанный.

Минусы:

Время. Job может пройти поздно, когда потребители уже прочитали плохие данные.
Ресурсы. Если запускать реже (раз в сутки вместо раза в час), за раз придётся обработать больше данных.
Sampling экономит ресурсы, но можно пропустить важное.

Online Observer

Проблема. Аналитики нашли неверный формат в поле zip code. Offline Observer тоже нашёл его, но запускался раз в неделю. Проблему надо видеть сразу.

Как работает:

Observation job становится частью пайплайна. Метрики готовы сразу после записи данных.
В batch есть два варианта: Parallel Split (observation идёт параллельно с загрузкой) и Local Sequencer (observation идёт после загрузки).
В streaming логика наблюдения встраивается в сам job. Отдельным шагом её не запустить.
В Spark можно использовать accumulators. Они считают невалидные строки и максимальные offsets за один проход, без второго запроса к данным.
В Airflow добавьте последнюю задачу с trigger_rule=all_done. Тогда сбой наблюдения не пометит весь пайплайн как проваленный.

Минусы:

Лишняя задержка в Local Sequencer: пайплайн заканчивается позже.
Parallel Split может увидеть частично валидный набор. Например, база не приняла формат даты, а observation читает данные до загрузки и не видит проблему. Тогда наблюдайте за тем, что видят потребители.
В streaming ошибка или нехватка памяти в наблюдении влияет на весь job. Можно брать sampling, но часть инсайтов будет потеряна.

	Offline Observer	Online Observer
Где живёт	Отдельный job	Внутри пайплайна
Влияет на основной job	Нет	Да
Скорость обратной связи	Медленная	Быстрая
Что выбирать	Нужна изоляция	Нужна скорость

Ключевые термины

Термин	Значение
`AWAP`	Audit-Write-Audit-Publish: аудит входа, запись, аудит результата, публикация
`audit`	Проверка данных, которая может остановить пайплайн
`data dispatching`	Валидные записи идут дальше, невалидные в отдельное место
`constraint`	Правило на уровне базы или формата: тип, `NOT NULL`, `CHECK`, integrity
`Schema Registry`	Сервис, который хранит версии схем и проверяет совместимость
`backward compatibility`	Новая схема читает старые данные
`forward compatibility`	Старая схема читает новые данные
`transitive`	Совместимость проверяется со всеми прошлыми версиями, а не только с соседней
`grace period`	Срок, когда живут и старое, и новое поле
`data observability`	Наблюдение за свойствами данных без блокировки пайплайна

Что это значит на практике

Добавьте в пайплайн два аудита: дешёвый на входе и содержательный на выходе. В dbt это tests и контракты моделей.
Не публикуйте таблицу, пока аудит не прошёл. Пишите сначала в staging.
Используйте NOT NULL, CHECK и типы в таблицах. Это самая дешёвая защита.
Для Kafka включите Schema Registry с режимом совместимости. Без него один продюсер может уронить всех.
Переименование колонки делайте в три шага: добавить новую, дать срок, удалить старую. Это работает и для dbt моделей.
Не считайте аудит финальной защитой. Добавьте observer и обновляйте правила по его метрикам.
Не каждая тревога аудита критична. Для спорных случаев используйте alert, а не fail.

Вопросы для самопроверки

Чем отличаются первый и второй аудит в AWAP? Почему не стоит повторять одну проверку в обоих?
Что такое backward и forward compatibility? Чем transitive режим отличается от nontransitive?
Как безопасно переименовать колонку, если у неё есть потребители?
Когда выбрать Offline Observer, а когда Online Observer?

Предыдущая: Паттерны хранения данных. Следующая: Паттерны наблюдаемости данных.