Идемпотентность загрузки данных: безопасные повторы

Прямой ответ

Идемпотентная загрузка даёт тот же результат при повторном выполнении, поэтому повтор после сбоя не создаёт дубликатов. Повтор — нормальная ситуация: загрузка прерывается по таймауту, перезапускается планировщиком либо выполняется вручную после исправления.

CREATE TABLE dwh.events (
    event_id   bigint NOT NULL,
    batch_id   text   NOT NULL,
    event_ts   timestamptz NOT NULL,
    payload    jsonb NOT NULL,
    PRIMARY KEY (event_id)
) PARTITION BY RANGE (event_ts);

-- Повтор партии не создаёт дубликатов
INSERT INTO dwh.events
SELECT event_id, :batch_id, event_ts, payload
FROM staging.events
WHERE batch_id = :batch_id
ON CONFLICT (event_id) DO NOTHING;

Уникальный ключ по идентификатору события исключает дубликат независимо от числа запусков. О применении upsert — в статье про upsert в витрине.

Идентификатор партии, перезапись раздела и определение повтора

Три механизма дополняют уникальность ключа и применяются в разных ситуациях.

Идентификатор партии. Каждая загрузка помечается идентификатором, поэтому данные конкретной партии различимы. Повтор партии заменяет её данные, а не добавляет к существующим.

Перезапись раздела. Данные размещаются по разделам за период, поэтому повторная загрузка периода перезаписывает раздел целиком. Механизм применим при партиционировании таблицы; о нём — в статье про секционирование таблиц.

Определение повтора. Повтор определяется по идентификатору партии либо по интервалу загрузки. Без явного признака повтор неотличим от новой загрузки.

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

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

ON CONFLICT (order_id) DO UPDATE
SET amount_minor = EXCLUDED.amount_minor,
    updated_at   = EXCLUDED.updated_at
WHERE dwh.orders.updated_at < EXCLUDED.updated_at

Аналогичная задача в прикладном коде решается идемпотентными операциями; о подходе — в статье про идемпотентность API.

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

Атомарность и частичное состояние при сбое

Идемпотентность и атомарность решают разные задачи, поэтому обе требуются для безопасного повтора.

Идемпотентность означает, что повторное выполнение даёт тот же результат. Обеспечивается уникальностью ключа либо заменяемостью единицы загрузки.

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

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

Загрузка, атомарная, но не идемпотентная, при повторе удваивает данные внутри одной транзакции.

Практическое следствие: единица загрузки заменяется целиком в одной транзакции, поэтому повтор не создаёт дубликатов и читатель не видит частичное состояние.

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

Практика: обеспечение идемпотентности загрузки

1. Определите единицу загрузки. Партия, интервал либо раздел — единица, которая заменяется целиком при повторе.

2. Обеспечьте уникальность ключа. Уникальный индекс по идентификатору записи исключает дубликат независимо от числа запусков.

3. Помечайте данные идентификатором партии. Признак партии позволяет различить повтор и новую загрузку.

4. Используйте перезапись раздела для данных за период. Повторная загрузка периода заменяет раздел, а не добавляет к нему.

5. Учитывайте порядок применения. Условие обновления по отметке изменения исключает перезапись актуальных данных старыми.

6. Проверяйте отсутствие дубликатов. Сравнение числа строк и уникальных ключей выявляет нарушение идемпотентности.

7. Логируйте выполненные загрузки. Записи о партиях и интервалах позволяют определить, что уже загружено.

Рекомендации:

1. Различайте идемпотентность и атомарность. Идемпотентность означает одинаковый результат при повторе, атомарность — отсутствие частичного состояния.
2. Не полагайтесь на отсутствие повторов. Планировщик перезапускает задачу после сбоя, поэтому повтор происходит без участия оператора.
3. Учитывайте стоимость проверки. Подсчёт уникальных ключей по всему объёму требует чтения всех данных, поэтому выполняется по загруженной части.
4. Фиксируйте результат загрузки. Число обработанных строк и признак завершения позволяют определить, требуется ли повтор.
5. Проверяйте идемпотентность повторным запуском. Тест с двойным выполнением подтверждает отсутствие дубликатов.

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

Контекст: после сбоя загрузка была запущена повторно, число строк в таблице увеличилось вдвое.

1. Диагностика. Загрузка выполнялась вставкой без условия конфликта, уникальный ключ по идентификатору события отсутствовал. Повторный запуск добавил те же записи, поскольку механизм не различал повтор и новую загрузку.
2. Причина. Идемпотентность не обеспечивалась: единица загрузки не была определена, данные не помечались идентификатором партии, поэтому повтор оказался неотличим от новой загрузки.
3. Действия. (а) Добавили уникальный ключ по идентификатору события; (б) ввели идентификатор партии и поле в таблице, чтобы различать запуски; (в) перешли на вставку с условием конфликта, чтобы повтор обновлял существующие строки; (г) удалили дубликаты, оставив по одной записи на идентификатор, и пересчитали производные витрины; (д) добавили проверку числа строк против уникальных ключей после каждой загрузки.
4. Результат. Повторный запуск загрузки не изменяет число строк, а расхождение обнаруживается проверкой.

Вывод: загрузка без уникального ключа и признака партии удваивает данные при повторе. Уникальность ключа в сочетании с идентификатором партии делает повтор безопасным.

FAQ

Что делает загрузку идемпотентной? Уникальный ключ с обработкой конфликта либо заменяемость единицы загрузки: партии, интервала либо раздела.

Как отличить повтор от новой загрузки? По идентификатору партии либо интервалу загрузки. Без явного признака повтор неотличим от новой загрузки.

Может ли повтор перезаписать актуальные данные старыми? Да, если порядок применения не учитывается. Условие обновления по отметке изменения исключает перезапись.

Как проверить идемпотентность? Повторным запуском загрузки с тем же набором данных: число строк и значения не должны измениться.

Источники: PostgreSQL INSERT, table partitioning, Spark SQL data sources — ссылки в шапке.