Airflow retry вызвал внешний API дважды: как защитить операцию

Задача Airflow отправила внешнему API запрос на создание выгрузки. API её создал, но соединение оборвалось до ответа. Airflow пометил попытку неудачной и повторил задачу. Теперь у провайдера две выгрузки, а в логе только один успешный ответ. Retry умеет запускать код заново; он не знает, состоялось ли внешнее побочное действие.

Для безопасного повтора нужен устойчивый идентификатор операции. Например, для ежедневного экспорта выберите ключ из имени процесса и границ data interval: orders:2026-04-01:2026-04-02. Обе попытки одного run передают один и тот же ключ, следующий день получает другой. Если API поддерживает idempotency key и сохраняет результат первого запроса, повтор вернёт тот же экспорт вместо создания второго.

Если API не знает idempotency key

Перед созданием проверьте, существует ли экспорт с устойчивым внешним идентификатором. После успешного создания сохраняйте ID ответа в надёжном хранилище, доступном следующей попытке. Но схема «проверить, затем создать» сама по себе не атомарна: две параллельные попытки могут одновременно увидеть отсутствие объекта. По возможности используйте уникальный ключ или операцию upsert на стороне API.

Хранить признак только в локальном файле worker нельзя: retry может попасть на другой worker. Одного XCom также недостаточно, если задача оборвалась до записи результата. Источник истины для операции должен переживать сбой процесса и позволять по ключу узнать, что уже произошло.

Сравните это с загрузкой в таблицу: задача может сначала удалить и заново заполнить партицию report_date в транзакции или сделать merge по устойчивому ключу строки. Тогда повтор того же интервала возвращает тот же итог. Простое INSERT без уникальности добавит дубли при retry и при ручном перезапуске.

Где граница гарантии

Идемпотентный ключ защищает только в пределах правил и срока хранения ключей конкретного API. Если провайдер забывает ключ через сутки, недельный backfill может снова создать объект. Узнайте TTL и поведение при том же ключе с другим телом запроса. Не скрывайте такую разницу: один ключ должен означать одну и ту же бизнес-операцию.

В журнале сохраняйте идентификатор DAG run, номер попытки, ключ операции и внешний ID. Тогда после сбоя можно установить, был ли ответ потерян после фактического создания. При ручном clear задачи тот же ключ должен применяться осознанно: иногда нужно повторить прежнюю операцию, иногда создать новую версию с явным новым ключом.

Попробуйте сформулировать требуемое свойство на одной строке: повтор задачи для того же интервала и того же входного состояния должен оставить один логический результат. Тогда проще проверить реализацию. Два физических HTTP-запроса допустимы, если API возвращает одну и ту же выгрузку; два разных экспорта под одним и тем же интервалом — уже нарушение. Для задач, которые отправляют письмо или запускают платный внешний процесс, цена такого нарушения особенно заметна.

Если после запроса связь оборвалась, следующий шаг не должен автоматически означать «создать снова». Сначала запросите состояние операции по устойчивому ключу, если API даёт такой метод. Если операции ещё нет, создайте её. Если она выполняется, дождитесь завершения. Если завершилась, используйте прежний результат. Эта схема может потребовать отдельной задачи ожидания, чтобы не держать worker во время долгой обработки на стороне провайдера.

При несовместимом изменении входных данных решите, будет ли это новая версия экспорта. Использование прежнего ключа с другим телом запроса может вызвать ошибку у API или вернуть старый результат. Добавляйте версию только осознанно: если в ключ включить номер попытки Airflow, каждый retry снова станет уникальной бизнес-операцией и вся защита исчезнет.

В курсе Data Engineering Fast‑Track Airflow организует загрузку, проверки и backfill. Откройте программу, чтобы строить DAG, который выдерживает повтор задачи без размножения результата.

Отдельный разбор дублей при Airflow backfill помогает перенести это правило на исторические интервалы.