Data Engineering Design Patterns, глава 10: Паттерны наблюдаемости данных

Главная мысль: Проверки качества смотрят на сами данные. Этого мало. Нужно ещё видеть, что пайплайн работает вовремя, и знать, откуда пришли данные. Для этого есть два столпа: детекторы (detection) и трекеры (tracking).

Пример из книги. Паттерн Audit-Write-Audit-Publish (AWAP) хорошо защищает от плохих данных. Но если сам AWAP-job не запустился из-за сбоя выше по потоку, вы об этом не узнаете. Наблюдаемость закрывает такие дыры.

Паттерны главы

Паттерн	Когда применять
`Flow Interruption Detector`	Данные перестали поступать, но job не упал
`Skew Detector`	Нужно убедиться, что объём данных не сильно изменился
`Lag Detector`	Нужно знать, как сильно consumer отстаёт от producer
`SLA Misses Detector`	Job должен завершаться за заданное время
`Dataset Tracker`	Нужно знать, какие датасеты и команды связаны между собой
`Fine-Grained Tracker`	Нужно знать происхождение отдельных колонок и строк

Детекторы данных (Data Detectors)

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

Flow Interruption Detector

Проблема. Streaming job синхронизирует данные в object store. Семь месяцев всё работало. Потом job начал читать записи, но не писать их. Он не упал, поэтому никто не заметил. Заметили потребители, когда пожаловались на отсутствие данных.

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

Для непрерывного потока (continuous delivery) ждём хотя бы одну запись за единицу времени. Если за минуту записей нет, шлём alert.
Для нерегулярного потока (irregular delivery) смотрим на окно. Alert идёт, если пауза длиннее допустимого окна. Так меньше ложных срабатываний и меньше alarm fatigue.
Для batch и таблиц смотрим на свежесть. Есть три слоя: metadata (время последнего изменения), data (колонка с временем изменения или сравнение количества строк), storage (время записи последнего файла).

Пример на Prometheus для Kafka: считаем сообщения в минуту и шлём alert, когда значение равно нулю.

sum without(instance)(rate(
  kafka_server_brokertopicmetrics_messagesin_total{topic="visits"}[1m]))

Для PostgreSQL можно включить track_commit_timestamp и читать время последнего commit через pg_xact_commit_timestamp. Producer тоже может сам отправить в Prometheus метрику «время последней записи».

Минусы:

Трудно выбрать порог. Одна запись в минуту может быть слишком мало или слишком много. Порог по истории даёт ложные срабатывания при пиках, например при маркетинговой акции.
Metadata может быть недоступна. Также время изменения меняется при смене схемы, хотя новых данных нет.
На уровне storage compaction создаёт новые файлы, но данные не новые. Это ложный сигнал активности.

Skew Detector

Проблема. Потребители жалуются, что batch job обработал неполный датасет. Job сработал правильно. Но у поставщика был сбой, и данных пришло вдвое меньше.

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

Выберите окно сравнения. Например, сегодня против вчера.
Задайте допуск (tolerance). Например, плюс-минус 50% объёма. Допуск берут из истории или спрашивают у бизнеса.
Посчитайте перекос. Первый способ: разница в процентах между окнами. Второй способ: отношение стандартного отклонения к среднему, STDDEV(x) / AVG(x). Он подходит для партиций, например в Kafka или PostgreSQL.

Пример для партиций таблицы PostgreSQL. Имена таблиц выдуманы.

SELECT STDDEV(n_live_tup) / AVG(n_live_tup) * 100 AS skew_pct
FROM pg_catalog.pg_stat_user_tables
WHERE relname LIKE 'visits_part_%';

Паттерн удобно ставить в первый audit-шаг AWAP. Он остановит обработку неполного датасета.

Минусы:

Сезонность. Рекламная кампания или лето и зима меняют объём. Одно правило для всех периодов не работает.
Нужна коммуникация с другими отделами. Иначе будут ложные срабатывания.
Fatality loop. Если вчерашний датасет был неполным и в три раза меньше, сегодняшний полный датасет выглядит как перекос. Решение: чинить данные у поставщика. Или сравнивать с последним успешным запуском, а не со вчерашним.

В Airflow это функция в PythonOperator. Она сравнивает размер текущего и прошлого файла и падает, если отношение вне диапазона 0.5-1.5.

Детекторы времени (Time Detectors)

Вторая категория смотрит на время и задержки.

Lag Detector

Проблема. Streaming job получил на 30% больше данных, чем раньше. Вы пропустили письмо об этом. Потребители жалуются на медленную доставку. Чтобы решить проблему с масштабированием, сначала нужно измерять, как быстро consumer читает.

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

Выберите единицу lag. Для Kafka это offset или время записи. Для Delta Lake это номер версии (commit). Для таблицы с партициями по времени это timestamp партиции.
Посчитайте разницу: lag = last_available_unit - last_processed_unit.
Для партиционированных источников агрегируйте lag по партициям. MAX покажет худший случай. Перцентили (P90, P95) покажут, как живёт большинство партиций. Лучше использовать оба.

Ловушка среднего. Допустим, lag по семи партициям равен 10, 5, 30, 2, 3, 5 и 3 секунды. Среднее равно 8 секундам. Но P90 равен 18 секундам. По среднему кажется, что всё хорошо, хотя это не так.

Минус: data skew. Если одна партиция получает больше нагрузки, MAX будет плохим. Но это не значит, что consumer плохой. Лучше исправить распределение данных.

Как собрать метрику:

Spark Structured Streaming: StreamingQueryListener читает offsets из onQueryProgress. Затем считает разницу и отправляет в Prometheus через Pushgateway.
Delta Lake: запустите job с триггером availableNow. Из прогресса запроса берите последнюю прочитанную версию. Последнюю записанную версию даёт DESCRIBE HISTORY.
Alert настройте на разницу между этими двумя версиями.

SLA Misses Detector

Проблема. Batch job стартует в 6:00 и должен закончиться за 40 минут. Потребители ждут данные до 8:00. Вы хотите узнавать о превышении 40 минут сразу.

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

Batch: время окончания минус время старта. Если разница больше порога, это SLA miss и нужен alert.
Streaming с microbatch или окнами: тот же расчёт на каждую итерацию.
Streaming без окон: для каждой записи считаем разницу между временем чтения и временем записи в выход. Потом агрегируем через MAX или перцентиль. Метрики можно собирать через Online Observer или Offline Observer из главы 9.

Lag и SLA не взаимозаменяемы. Пример: consumer ограничен по throughput и всегда обрабатывает одинаковый объём. SLA соблюдается, но lag растёт. Обратный пример: у batch job нет lag на старте, но он может долго работать и нарушить SLA.

Минусы:

В batch всё просто. В streaming труднее из-за семантики прихода данных.
Late data и event time. SLA по processing time показывает скорость вашего job. SLA по event time включает задержку поставщика, например при потере сети. Эти два SLA лучше разделять и мерить оба.
В Airflow параметр sla считается от старта всего DAG run, а не от старта задачи. Если запуск назначен на 08:00 и задача с sla 10 секунд стартует в 08:00:10, она уже опоздала. В Airflow 3 параметр sla убрали. Вместо него в Airflow 3.1 появились Deadline Alerts.

В Flink книга показывает такой способ. Первый job пишет в запись время начала обработки. Второй job читает это поле и время записи в Kafka. Он считает разницу и выдаёт перцентили по окнам в одну минуту.

Data Lineage

Что делать, если проблема не у вас? Например, SLA по event time нарушила другая команда. Тогда нужно знать, к кому идти. Для этого строят граф происхождения датасетов: кто из чего сделан.

Dataset Tracker

Проблема. Ваш batch job падает из-за изменения типа поля. Поставщик не знает об этом, потому что сам обрабатывает данные другой команды. Вы хотите понять цепочку зависимостей и найти, где появилась проблема.

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

Строим граф зависимостей между контейнерами данных: таблицами, папками, топиками, очередями. У каждого узла есть команда-владелец.
Вариант 1, управляемый сервис. Он сам строит lineage. Примеры: Unity Catalog в Databricks и Dataplex в GCP.
Вариант 2, свой lineage. Входы и выходы собирают на уровне оркестратора (Airflow с OpenLineage) или на уровне базы (анализ выполненных SQL-запросов, FROM и JOIN).
Нужны ещё API и UI для просмотра графа. Например, OpenLineage и Marquez.

Минусы:

Vendor lock. Управляемые решения часто видят только свои сервисы. Данные в чужом облаке видны частично.
Custom work. Оркестратор знает входы и выходы только для стандартных операторов. Для своих задач нужно писать логику самим.

Для Spark подключите listener OpenLineage при создании SparkSession. Для Airflow задайте OPENLINEAGE_URL и установите provider-пакет.

Fine-Grained Tracker

Проблема. Денормализованная таблица выросла до 30+ колонок. Новые коллеги спрашивают, из каких колонок исходных таблиц собрана каждая колонка. Dataset Tracker на это не отвечает.

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

Column-level lineage. Некоторые платформы дают его из коробки: system.access.column_lineage в Databricks, Purview в Azure. Свой вариант: разбор плана выполнения запроса и поиск всех входных колонок для каждой выходной.
Row-level lineage. Добавляем в запись информацию о job, который её создал. Это делают через дополнительную колонку или заголовки (headers) сообщения Kafka. Так можно хранить версию job, имя job, номер batch и lineage родителя.

Пример для CONCAT(u.first_name, d.delivery_address). Выходная колонка зависит от users.first_name и addresses.delivery_address.

Минусы:

Custom code. Если трансформация идёт через свою функцию, lineage-инструмент видит чёрный ящик и не поймёт логику.
Row-level lineage не интегрируется с обычными инструментами визуализации. Нужен свой слой запросов.
Evolution. Логика колонок меняется. Инструмент должен хранить историю, иначе вы получите неверное происхождение колонки.

Послесловие автора

Автор пишет, что инструменты будут меняться. SQL, Spark, форматы таблиц и Kafka-совместимые брокеры могут уйти. Но задачи останутся теми же:

привозить данные, непрерывно или по расписанию;
превращать сырые данные в ценные;
проверять, что всё работает правильно.

Паттерны книги дают устойчивую основу в меняющемся мире.

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

Термин	Значение
`Observability`	Видимость состояния данных и пайплайнов: мониторинг и alerting
`Alarm fatigue`	Усталость от слишком частых и ложных alert
`Data skew`	Неравномерный объём данных между партициями или запусками
`Lag`	Отставание consumer от последней доступной записи
`SLA`	Максимальное допустимое время выполнения job
`Percentile (P90, P95)`	Значение, в пределах которого лежит 90% или 95% измерений
`Event time` / `Processing time`	Время создания записи / время её обработки
`Data lineage`	Происхождение данных и связи между датасетами
`OpenLineage`	Открытый стандарт и API для сбора lineage
`Fatality loop`	Один плохой запуск ломает сравнение для следующих

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

Добавьте alert на отсутствие новых данных. Job без ошибок не значит, что данные доходят.
Используйте P90, P95 и MAX, а не среднее. Среднее прячет проблемные партиции.
Сравнивайте объём данных с допуском и учитывайте сезонность. Договоритесь с бизнесом о ложных срабатываниях.
Ставьте Skew Detector в первый audit-шаг AWAP и в проверки dbt-пайплайна перед публикацией.
Измеряйте SLA отдельно по processing time и по event time. Так вы поймёте, чья это проблема.
Подключите OpenLineage к Airflow и Spark. Lineage быстрее покажет, к какой команде идти при инциденте.
Для критичных датасетов храните в записях версию job и имя job. Это помогает при отладке.

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

Чем continuous и irregular delivery различаются для Flow Interruption Detector? Почему для второго нужны окна?
Что такое fatality loop в Skew Detector и как его избежать?
Почему для lag лучше использовать перцентили, а не среднее?
Чем SLA по processing time отличается от SLA по event time? Зачем мерить оба?

Предыдущая: Паттерны качества данных.