Data Engineering Design Patterns, глава 3: Паттерны обработки ошибок

Главная мысль: Ошибки в пайплайнах неизбежны: плохие записи, дубли, опоздавшие данные, сбои задач. Паттерны главы не убирают ошибки. Они делают так, чтобы пайплайн жил с ними и данные не портились.

Ошибки бывают двух типов:

Transient: временные. Они проходят сами. Пример: база недоступна несколько секунд. Помогает retry.
Nontransient: сами не проходят. Пример: poison pill message, запись, которую нельзя обработать. Нужно вмешательство человека или специальный паттерн.

Автоматические retry создают новую проблему: дубли. Поэтому глава 3 заканчивается выводом: нужна идемпотентность. Это тема главы 4.

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

Паттерн	Когда применять
`Dead-Letter`	Одна плохая запись не должна останавливать весь job
`Windowed Deduplicator`	Источник дублирует записи из-за retry продюсера
`Late Data Detector`	Нужно понять, что событие пришло слишком поздно
`Static Late Data Integrator`	Опоздавшие данные приходят в известном окне (например, до 15 дней)
`Dynamic Late Data Integrator`	Окно опоздания неизвестно. Хотите пересчитать только затронутые партиции
`Filter Interceptor`	Нужно знать, какой фильтр отбросил сколько строк
`Checkpointer`	Стриминговый job должен продолжить с места остановки после сбоя

Необработанные записи

`Dead-Letter`

Проблема. Streaming job читает события из Kafka. Продюсер начал слать битые записи. Job падает. Вы вручную двигаете offset. Так не может продолжаться.

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

Найдите места, где job может упасть: функция map, парсинг, преобразование.
Оберните их в try-catch (или if-else, если фреймворк возвращает NULL вместо ошибки).
Плохую запись отправьте в отдельное хранилище: dead-letter storage. Добавьте метаданные: текст ошибки, имя job, версию, время.
Выберите хранилище с высокой доступностью и простым мониторингом: object store или брокер. Следите за числом записей в нём.
По желанию добавьте replay pipeline. Он возвращает исправленные записи в основной поток.

Паттерн работает и в batch. Там job записывает в dead-letter сразу пачку плохих строк.

Компромиссы:

Snowball backfilling. После replay записи попадают в уже обработанные партиции. Потребители должны пересчитать их. Их потребители тоже. Пересчёт идёт по цепочке.
Как отличить replay-записи. Добавьте колонку вроде was_dead_lettered или метаданные: job, версия, время replay.
Порядок и согласованность. Replay ломает порядок. Пример: записи 10:00, 10:01, 10:02, а запись 10:01 вернулась позже. Сессии по окну неактивности тоже могут закрыться неверно.
Error-safe функции. Они возвращают NULL вместо исключения. Нужно сравнивать вход и выход: вход не NULL, а выход NULL значит ошибку. В SQL код становится длинным.
Ошибка или сбой. Паттерн скрывает проблемы. Добавьте алерт. Если плохих записей слишком много, job должен остановиться.

Идея на SQL (данные придуманы):

-- TRY_CAST возвращает NULL вместо ошибки.
-- Вход не NULL, а выход NULL: значит, запись битая.
SELECT *, NOT (raw_price IS NOT NULL AND price IS NULL) AS is_valid
FROM (SELECT order_id, raw_price, TRY_CAST(raw_price AS DECIMAL(10, 2)) AS price
      FROM orders_raw);
-- дальше: is_valid = true пишем в основную таблицу, false в dead-letter таблицу

В Flink это делают через side output. В Spark пишут два результата из одного кэшированного (persist) датафрейма.

Дубликаты

`Windowed Deduplicator`

Проблема. Batch job читает события, которые стриминг-слой синхронизировал в object store. Из-за retry продюсеров в данных есть дубли. Пользователям нужна обработка «ровно один раз».

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

Выберите атрибуты, которые делают запись уникальной (visit_id или пара visit_id + visit_time).
В batch область дедупликации это текущий набор данных. Используйте DISTINCT или ROW_NUMBER() ... = 1. В Spark есть dropDuplicates.
В streaming данных бесконечно много. Поэтому дедупликация идёт в окне по времени. Job хранит уже виденные ключи в state store.
Старые ключи удаляются по watermark. Так состояние не растёт бесконечно.

Три вида state store:

Вид	Скорость	Риск
Local (в памяти)	Самый быстрый	Состояние пропадёт при сбое
Local с fault tolerance	Быстрый, но есть запись на диск	Чем чаще запись, тем медленнее job
Remote	Надёжный	Выше latency и стоимость

Компромиссы:

Память против времени. Короткое окно пропустит часть дублей. Длинное окно требует больше состояния.
Дедупликация не равна exactly-once delivery. При retry после сбоя запись может обработаться второй раз. Для настоящего exactly-once нужны паттерны идемпотентности (глава 4).

# Spark Structured Streaming: ключ живёт 10 минут
(events.withWatermark("visit_time", "10 minutes")
       .dropDuplicates(["visit_id", "visit_time"]))

Опоздавшие данные

Важно различать два времени:

event time: когда событие произошло.
processing time: когда пайплайн его увидел. Оно никогда не опаздывает.

`Late Data Detector`

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

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

Нужен атрибут времени события.
Для каждой партиции отслеживайте прогресс по event time. Используйте MAX: значение должно только расти.
Для всех партиций вместе выберите стратегию:
- MIN: ждём самую медленную партицию. Меньше потерь, больше буфер.
- MAX: идём за самой быстрой. Меньше буфер, больше пропусков.
- Комбинация: например, MAX внутри источника и MIN между источниками.
Вычислите watermark = MAX(event time) - allowed lateness. Событие раньше watermark считается опоздавшим.
Новый кандидат в watermark меньше текущего игнорируйте. Watermark не идёт назад.

Пример: допустимое опоздание 30 минут. Пришли события 10:00, 10:05, 10:06. Watermark станет 9:36. Затем пришло 9:20 и 9:31. Они раньше 9:36, значит это опоздавшие.

Компромиссы:

Spark Structured Streaming сам отбрасывает опоздавшие события. Достать их оттуда трудно. Во Flink это проще: ProcessFunction видит текущий watermark и может отправить запись в side output.
MIN на уровне партиции опасен: возможен эффект stuck in the past. Watermark может не сдвинуться никогда, а состояние растёт. Ещё возможен цикл «закрыл, открыл, закрыл» для уже выданных агрегатов.
MAX при перекосе слишком агрессивен. Если 4 из 5 источников опоздали на 40 минут, их записи будут считаться поздними. Помогает мониторинг и повторная загрузка опоздавших.

`Static Late Data Integrator`

Проблема. Daily batch job считает статистику. Данные считаются приблизительными 15 дней. Job сейчас обрабатывает только текущий день и теряет опоздавшие записи. Хочется включить их без 15 отдельных запусков.

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

Задайте статическое окно lookback (например, 14 дней). Каждый запуск пересчитывает текущий день плюс все дни в окне.
Поместите интеграцию в основной пайплайн. Порядок такой:
- Stateful пайплайн: сначала опоздавшие данные, потом текущий день.
- Stateless пайплайн: любой порядок, в том числе параллельно.
В Airflow удобно использовать Dynamic Task Mapping: задачи по дням создаются через expand.

Совет: партиционирование по processing time решает задачу только для вас. Данные всё равно опоздают для тех, кто читает по event time.

Компромиссы:

Snowball backfilling: потребители тоже должны пересчитать партиции. Предупредите их.
Запуски перекрываются. При окне 4 дня запуски 10, 11 и 12 октября пересчитывают одни и те же дни. Перезапускайте только последний запуск.
Backfill-задачи должны быть внутри основного пайплайна. Отдельный пайплайн на каждую партицию даёт те же перекрытия.
Ресурсы тратятся зря, если в окне нет опоздавших данных. Можно добавить контрольную задачу.
Нужны партиции по времени.

`Dynamic Late Data Integrator`

Проблема. Окна в 15 дней мало. Нужно включать опоздавшие данные любой давности, но пересчитывать только затронутые партиции.

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

Заведите state table: партиция, last processed time, last update time, флаг is_processed.
Партиция требует backfill, если last update time > last processed time и она старше текущей.
После обработки обновляйте last processed time.
last update time берите у хранилища. В BigQuery это INFORMATION_SCHEMA.PARTITIONS. В Iceberg есть метаданные партиций. Для Delta Lake считайте по версиям таблицы через DeltaLog.
Если можно выделить затронутые сущности, перезаписывайте не всю партицию, а только их часть.

SELECT partition
FROM state_table
WHERE last_update_time > last_processed_time
  AND partition < current_partition
  AND is_processed = false;

Компромиссы:

Concurrency. Параллельные запуски могут взять одну партицию дважды. Решение: флаг is_processed и depends_on_past=True на ключевых задачах Airflow. Если такая задача упадёт, пайплайн зависнет и нужно ручное вмешательство.
Очень старые данные в stateful пайплайне. Придётся пересчитать всю цепочку запусков. Поэтому всё равно нужен максимальный допустимый возраст опоздания.
Сложность. Нужно строить backfill пайплайны динамически. Метаданные хранилища не всегда доступны.

Фильтрация

`Filter Interceptor`

Проблема. После релиза доля отфильтрованных строк выросла с 15% до 90%. Это баг или изменились данные? План запроса не поможет: движок объединяет условия фильтра в одно.

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

В коде с API (Scala, PySpark) оберните каждое условие фильтра. При false увеличивайте счётчик (в Spark это accumulator).
В конце job соберите значения счётчиков.
В SQL вынесите условия в колонки подзапроса, например status_flag со значениями null_type, short_type. Затем посчитайте строки по флагам и отфильтруйте.

Компромиссы:

Небольшие накладные расходы. В SQL больше, если нужна временная таблица.
Декларативный SQL здесь слабее, чем программный API. Используйте то, что проще.
В streaming фильтр становится stateful. Нужны границы по времени (окна), иначе статистика теряет смысл.

Отказоустойчивость

`Checkpointer`

Проблема. Streaming job считает уникальные визиты в 10-минутных окнах. После сбоя он не должен читать всё с начала.

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

Job сохраняет последнюю прочитанную позицию (offset) и состояние в надёжное хранилище.
Два варианта:
- Через фреймворк: Spark Structured Streaming и Flink пишут метаданные в object store.
- Через SDK хранилища: Kafka (__consumer_offsets), Kinesis KCL (таблица DynamoDB).
Два способа записи: по конфигурации (частота checkpoint) или явным вызовом commit в вашем коде.

# Spark: путь для checkpoint
(stream.writeStream
       .option("checkpointLocation", "/data/checkpoint")
       .start())

Компромиссы:

Гарантия против latency. Частые checkpoint замедляют job. Редкие означают больше повторной работы после сбоя. Состояние обычно намного больше, чем offset.
Это не exactly-once. Параллельные асинхронные задачи могут упасть до checkpoint. Для exactly-once нужны паттерны идемпотентности.
Момент записи важен: checkpoint после обработки даёт at least once (возможны дубли). Checkpoint до обработки даёт at most once (возможна потеря).

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

Термин	Значение
Transient error	Временная ошибка. Проходит сама, помогает retry
Nontransient error	Постоянная ошибка. Нужна ручная работа
Poison pill message	Запись, которую нельзя обработать. Останавливает job
Dead-letter storage	Отдельное хранилище для плохих записей
Replay pipeline	Пайплайн, который возвращает записи из dead-letter в поток
Snowball backfilling	Backfill одного пайплайна вызывает backfill всех потребителей
State store	Хранилище состояния streaming job
Event time / processing time	Время события / время обработки
Watermark	Граница `event time`. Раньше неё записи считаются опоздавшими
Allowed lateness	Допустимое опоздание при расчёте watermark
Lookback window	Окно в прошлое, которое пересчитывается при каждом запуске
Side output	Дополнительный выход job для особых записей
Checkpoint	Сохранённая позиция и состояние job
At least once / at most once	Возможны дубли / возможна потеря

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

Не давайте job падать из-за одной плохой записи. Пишите её в dead-letter и ставьте алерт на объём.
Если плохих записей много, остановите job. Иначе вы молча потеряете данные.
Продумайте replay до запуска. Спросите: кто ещё читает эти партиции и сможет ли он пересчитать?
Дедуплицируйте по ключу и по окну. Но не считайте это exactly-once. Делайте запись идемпотентной.
Всегда выбирайте allowed lateness осознанно. Мониторьте число опоздавших записей.
В Airflow используйте expand для окна lookback. Перезапускайте только последний запуск, а не все.
Считайте отфильтрованные строки по каждому условию. Это быстро показывает, баг в коде или в данных.
Всегда включайте checkpoint в streaming. Подберите частоту под допустимую задержку.

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

Чем отличаются transient и nontransient ошибки? Как Dead-Letter помогает со вторыми?
Почему Windowed Deduplicator не даёт настоящего exactly-once? Что делает watermark в этом паттерне?
Почему для watermark на уровне партиции берут MAX, а не MIN? Что такое stuck in the past?
Когда нужен Dynamic Late Data Integrator вместо Static? Как избежать двойного backfill при параллельных запусках?

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