Data Engineering Design Patterns, глава 4: Паттерны идемпотентности

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

Глава 3 учит обрабатывать ошибки. Но автоматический retry сам создаёт проблему. Задача может повторить запись, которая уже прошла. Тогда в таблице появятся дубликаты. Иногда их даже нельзя отличить друг от друга. Идемпотентность решает это.

Простой пример: abs(-1) == abs(abs(abs(-1))). Сколько раз ни вызывай функцию, ответ тот же. Для пайплайна это значит: результат не зависит от числа запусков.

Дубликаты не всегда можно убрать. Например, брокер без транзакций при retry может записать сообщение дважды. Тогда идемпотентная обработка помогает хотя бы опознать дубликаты.

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

Семейство	Паттерн	Когда применять
Overwriting	Fast Metadata Cleaner	Нужно быстро очистить большую таблицу, и БД умеет TRUNCATE или DROP
Overwriting	Data Overwrite	Метаданных нет (например, object store), но можно заменить данные
Updates	Merger	Приходит только часть данных (изменения), записи имеют стабильный ключ
Updates	Stateful Merger	Нужен безопасный backfill для Merger
Database	Keyed Idempotency	Стриминг или key-value база, ключ можно построить из неизменяемых полей
Database	Transactional Writer	Потребители не должны видеть частичную запись
Immutable dataset	Proxy	Данные нельзя менять или удалять, а читать нужно одну актуальную версию

Overwriting: перезапись данных

Самый простой путь к идемпотентности: удалить старые данные и записать новые. Проблема в стоимости. На больших таблицах DELETE медленный. Поэтому есть два паттерна: через метаданные и через данные.

Fast Metadata Cleaner

Проблема. Ежедневный job пишет сотни ГБ. Сначала он делает DELETE строк прошлого запуска, потом INSERT. Таблица растёт, DELETE становится всё медленнее.

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

• DELETE ищет строки и переписывает файлы. TRUNCATE и DROP TABLE работают только с метаданными. Поэтому они быстрые.
• Данные хранят в нескольких физических таблицах, например по одной на неделю. Сверху делают view, которая объединяет их.
• Оркестратор проверяет дату запуска. В первый день новой недели он создаёт таблицу, добавляет её в view и очищает. В другие дни сразу грузит данные.
• Для DROP порядок такой: убрать таблицу из view, удалить, создать заново, вернуть в view. Иначе читатель увидит ошибку.

Идея в том, что набор данных состоит из нескольких таблиц, а не из одной. Часть можно сбросить дёшево.

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

• Гранулярность идемпотентности. Если таблица недельная, backfill одного дня требует перезапуска всей недели. Для остальных дней можно перезапустить только шаг загрузки.
• Точечный backfill (один клиент, один поставщик) не работает. Метаданные относятся к таблице целиком.
• Лимиты хранилища. У BigQuery и Redshift есть лимит на число партиций и таблиц. Помогает шаг freezing: недельные таблицы позже склеивают в месячные или годовые.
• Нужна поддержка в БД. Паттерн работает в хранилищах, lakehouse и реляционных БД. В object store его трудно сделать.
• Нужен единый вход (view), чтобы пользователи не знали о внутренних таблицах.
• Schema evolution. Новое необязательное поле надо добавлять в старые таблицы отдельным пайплайном.

Data Overwrite

Проблема. Данные лежат в object store, партиции по event time. TRUNCATE и DROP там нет. Каждый backfill создаёт дубликаты.

Как работает. Паттерн заменяет сами данные. Способы зависят от стека:

• Framework: режим записи overwrite в Spark или Flink. Delta Lake умеет заменить только часть данных через replaceWhere.
• SQL, вариант 1: DELETE и затем INSERT. Он позволяет выбрать строки.
• SQL, вариант 2: INSERT OVERWRITE. Он короче, но заменяет всю таблицу или партицию без выбора строк.
• Команды загрузки: LOAD DATA OVERWRITE в BigQuery или флаг --replace=true. В других системах перед загрузкой делают TRUNCATE.

-- Заменяем содержимое таблицы новой версией (пример с выдуманными таблицами)
INSERT OVERWRITE INTO orders
SELECT * FROM orders_staging WHERE status = 'valid';

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

• Объём данных. Без партиционирования перезапись замедляется с ростом таблицы. Помогают партиции и другие приёмы хранения (глава 8).
• Нужен vacuum. DELETE может не стереть данные физически. В table formats, BigQuery и Snowflake старые блоки остаются из-за time travel. Они удаляются после retention или после vacuum.
• Режим overwrite в Spark не транзакционный. Всё зависит от формата. В Delta Lake удаление это новый commit, файлы остаются на месте.

Updates: обновление данных

Полная замена не подходит, если поставщик присылает только изменённые строки. Тогда новые строки нужно объединить со старыми.

Merger

Проблема. Изменения из Kafka (CDC) нужно применить к таблице Delta Lake. В таблице не должно быть дубликатов.

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

• Сначала определяют ключ слияния. Это одно поле (user_id) или несколько (visit_id и visit_time).
• Используют команду MERGE (она же UPSERT). Её поддерживают движки, table formats и хранилища.
• Три случая: нет в таблице, значит INSERT. Есть в таблице, значит UPDATE. Удаление.
• MERGE не умеет удалять то, чего нет во входе. Поэтому удаления передают как soft delete: поле is_deleted = true.
• В ветке INSERT тоже проверяют is_deleted = false. Иначе при первом запуске удалённая запись попадёт в таблицу и там останется.

MERGE INTO customers AS t
USING customers_input AS i ON t.customer_id = i.customer_id
WHEN MATCHED AND i.is_deleted THEN DELETE
WHEN MATCHED THEN UPDATE SET name = i.name
WHEN NOT MATCHED AND NOT i.is_deleted THEN
  INSERT (customer_id, name) VALUES (i.customer_id, i.name);

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

• Уникальность. Нужен неизменяемый ключ. Без него MERGE вставит новую строку вместо обновления.
• I/O. MERGE работает с данными, а не с метаданными, поэтому он дороже. Современные движки сначала ищут нужные файлы по метаданным и пропускают лишние.
• Backfill ломает консистентность. Пайплайн стартует с последней версии таблицы. Часть строк, записанных после точки старта, отсутствует. Потребители временно видят другие данные, чем при обычном запуске.
• Если поставщик исправил ошибки, replay не нужен. Попросите набор с исправлениями и примените его как новый инкремент, если ключи не изменились.

Stateful Merger

Проблема. Через неделю нашли ошибку, и бизнес просит backfill. Но он хочет сначала откатить таблицу к последней корректной версии. Обычный Merger этого не умеет.

Как работает. К пайплайну добавляют таблицу состояния (state table) и два шага:

• В начале: восстановить таблицу, если нужно.
• В конце: записать в state table пару “время запуска и версия таблицы после MERGE”.
• Как понять режим. Берут версию, созданную прошлым запуском, и сравнивают с текущей версией таблицы. Равны: обычный запуск. Отличаются: backfill, нужен откат к версии прошлого запуска.
• Если версии для прошлого запуска нет, это первый запуск. Таблицу очищают TRUNCATE.
• Оркестратор часто сам знает, что идёт backfill. Тогда сравнение версий не нужно.

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

• Нужна версионность. Table formats (Delta Lake, Iceberg) её дают. Без них хранят все сырые данные в отдельной таблице с временем загрузки. При backfill удаляют строки с будущим временем и пересобирают таблицу (например, паттерном Windowed Deduplicator).
• vacuum и retention. Старые версии удаляются. Откат за пределы retention невозможен.
• Операции без изменения данных. Compaction не меняет данные, но создаёт новую версию. Если брать просто “прошлую версию”, можно потерять операции между двумя MERGE. Тогда берут версию текущего запуска и вычитают 1.

Database: идемпотентность силами БД

Если оркестрации мало или её нет (например, стриминг), часть работы делает база данных.

Keyed Idempotency

Проблема. Стриминговый job строит пользовательские сессии и пишет их в key-value хранилище. При retry не должно быть дубликатов.

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

• Ключ строят из неизменяемых полей. Тогда повторная запись просто перезаписывает ту же запись.
• Если есть готовый уникальный атрибут (user_id), берут его.
• Если нет, нужна комбинация. Ловушка: user_id и время первого события. Позднее событие поменяет “первое” время, и ключ изменится.
• Безопаснее брать append time: время физической записи в брокер. В Kafka это timestamp сообщения, в Kinesis это approximate arrival timestamp. В таблицах это ingestion time.
• Для batch ключом может быть дата запуска. Файл 2024_11_20 при повторе будет перезаписан.

-- Первая запись по пользователю по неизменяемому времени загрузки
SELECT user_id,
       FIRST_VALUE(visit_time) OVER (
         PARTITION BY user_id ORDER BY ingestion_time ASC, visit_time ASC
       ) AS first_visit
FROM visits;

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

• Зависит от БД. В Cassandra, ScyllaDB, HBase это работает сразу. В реляционной БД INSERT с тем же ключом даёт ошибку, нужен MERGE. Kafka хранит дубликаты с одним ключом, а убирает их позже через compaction.
• Изменяемый источник. Compaction может удалить первое событие, которое использовалось для ключа. После рестарта ключ изменится. Но и данные изменились, поэтому новый ключ допустим.
• В Kafka append time лучше брать из настройки брокера (log.message.timestamp.type). Время, заданное продюсером, менее надёжно.

Transactional Writer

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

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

• Три шага: открыть транзакцию (явно через BEGIN или неявно), записать данные, сделать commit.
• Пока нет commit, потребитель изменений не видит. При ошибке делают rollback или просто не делают commit.
• Для хранилищ (BigQuery, Redshift, Snowflake) транзакцию ведёт сама БД.
• Для распределённых job есть два варианта. Локальная транзакция у каждой задачи. Или одна транзакция на весь job: commit после завершения всех задач.
• Поддержка: Delta Lake, Iceberg, Hudi, Kafka, хранилища, PostgreSQL, MySQL и другие СУБД.

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

• Шаг commit. Данные становятся видимыми только после commit. Потребители ждут самую медленную задачу.
• Поддержка в движках неполная. Kafka-транзакции есть в Flink, но их нет в Spark.
• Идемпотентность только в рамках транзакции. При локальных транзакциях рестарт job перепишет данные уже закоммиченных задач. Backfill тоже добавит те же записи новой транзакцией.
• Читатели с уровнем read uncommitted могут увидеть незафиксированные записи (dirty reads).
• В Flink с EXACTLY_ONCE таймаут транзакции должен быть больше времени checkpoint. Иначе commit не пройдёт.

Immutable dataset: неизменяемые данные

Proxy

Проблема. Job каждый раз пишет полный набор и затирает старый. Юристы просят хранить все версии, но отдавать только последнюю.

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

• Каждую загрузку пишут в новое место: таблица с суффиксом версии или времени.
• После создания таблицы убирают права на запись. В object store для этого есть WORM: Object Lock в S3, immutability policies в Azure Blob, bucket locks в GCP.
• Читатели ходят через единую точку: view вида SELECT * FROM <последняя таблица>.
• Без view можно вести manifest file со списком файлов. Его создание лучше отделить от обработки данных.
• В Delta Lake, Iceberg и BigQuery есть time travel. Там отдельные таблицы не нужны, но retention часто ограничен (в BigQuery это 7 дней на момент написания книги).

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

• Нет view в БД. Тогда нужен manifest, и чтение усложняется.
• Неизменяемость нужно закрепить и в хранилище. Настройки оркестратора недостаточно. Нужны блокировки и запрет записи, значит нужна помощь инфраструктурной команды.
• Имя таблицы должно быть уникальным при каждом запуске, например из времени старта DAG. Иначе повторный запуск попробует перезаписать старую таблицу.
• Права пользователя job должны разрешать только создание таблиц, не удаление.

Как выбрать паттерн

Ситуация	Паттерн
Данные поставляются целиком, БД умеет TRUNCATE	Fast Metadata Cleaner
Данные поставляются целиком, метаданных нет	Data Overwrite
Приходят только изменения	Merger
Изменения приходят, и нужен консистентный backfill	Stateful Merger
Стриминг, key-value хранилище	Keyed Idempotency
Нельзя показывать частичные данные	Transactional Writer
Нужна неизменяемость и история	Proxy

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

Термин	Значение
Idempotency	Повторный запуск даёт тот же результат
Idempotency granularity	Единица, которую можно очистить и перезапустить (например, неделя)
Metadata operation	Операция над описанием данных (TRUNCATE, DROP), а не над файлами
MERGE / UPSERT	Вставка новых и обновление существующих строк одной командой
Soft delete	Удаление через флаг, например is_deleted
State table	Таблица с версиями данных по запускам пайплайна
Append time	Время записи сообщения в брокер. Не меняется при позднем событии
Commit / rollback	Фиксация или откат транзакции
WORM	Write once read many: запись один раз, чтение много раз
Vacuum	Физическое удаление устаревших файлов и версий

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

• Проектируйте каждый job так, чтобы его можно было запустить дважды. Retry и backfill в Airflow это норма.
• Для полных наборов выбирайте перезапись, для инкрементов выбирайте MERGE. Это самый простой путь.
• В dbt инкрементальные модели с unique_key это по сути Merger. Проверьте, что ключ неизменяемый.
• Не делайте ключ из значений, которые могут измениться. Пример: время первого события сессии. Позднее событие его поменяет. Лучше брать время загрузки.
• Если нужен откат при backfill, используйте time travel (Delta Lake, Iceberg) и таблицу состояния.
• Помните: Checkpointer из главы 3 не даёт exactly-once. Повторную запись делают безопасной паттерны этой главы.
• Для аудита и юридических требований храните версии и отдавайте последнюю через view.

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

1. Почему TRUNCATE быстрее DELETE и как это используется в Fast Metadata Cleaner?
2. Зачем в MERGE для Merger нужен soft delete и проверка is_deleted в ветке INSERT?
3. Почему при backfill Merger даёт временно неконсистентные данные и как это решает Stateful Merger?
4. Почему в Keyed Idempotency лучше брать append time, а не event time?

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