Backfill пересекается с текущей загрузкой: как согласовать владельца партиции

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

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

Почему одинаковый день ещё не означает одинаковый расчёт

У каждого кандидата зафиксируйте входной snapshot_id, версию преобразования и идентификатор попытки. Если пересчёт использует исправленные правила, он может иметь приоритет над расписанием. Если обычная загрузка включает более поздние события, приоритет может быть у неё. Это решение задаётся политикой обновления данных, а не длительностью выполнения. Случайный последний писатель не выражает ни один из этих смыслов.

Airflow позволяет управлять диапазоном backfill и его параллельностью. При этом max_active_runs для backfill применяется отдельно от ограничения DAG. Эти настройки полезны для нагрузки, но координация записи одной партиции остаётся задачей конвейера. Два разных DAG или внешний загрузчик тоже могут обращаться к той же таблице. Поэтому проверку владельца размещайте у операции публикации.

Добавим номер права на публикацию

Рассмотрим ситуацию, когда старая задача зависла, а право работы передали новой. Одной проверки owner в начале недостаточно: старая задача может проснуться и записать результат после новой. Монотонный номер выдачи права, иногда называемый fencing token, позволяет отвергнуть такой поздний commit. Проверять его должен сам слой записи, а не только код работника.

partition = "orders/2026-10-28"
registry = {partition: {"token": 0, "owner": None, "value": None}}
def acquire(key, owner):
    entry = registry[key]
    entry["token"] += 1
    entry["owner"] = owner
    return entry["token"]
def publish(key, owner, token, value):
    entry = registry[key]
    if (entry["owner"], entry["token"]) != (owner, token):
        return False
    entry["value"] = value
    return True
old = acquire(partition, "schedule")
new = acquire(partition, "backfill")
assert publish(partition, "backfill", new, "corrected-snapshot")
assert not publish(partition, "schedule", old, "old-snapshot")
print(registry[partition])
assert registry[partition]["value"] == "corrected-snapshot"

Пересчёт получил token 2 и опубликовал исправленный снимок. Поздняя запись расписания с token 1 отклонена. В этом примере acquire намеренно передаёт право без ожидания; решение о передаче уже принято оператором. В рабочем реестре понадобятся условное атомарное обновление, срок аренды и правило безопасной передачи. Обычный словарь показывает условие допуска, но не даёт межпроцессной блокировки.

Как организовать пересчёт без остановки всего конвейера

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

Готовьте результат в отдельном месте с идентификатором попытки. При публикации проверяйте актуальное право, полноту и версию входа, затем переключайте партицию средствами хранилища. После отказа не пытайтесь повторить commit бесконечно с тем же token. Работник должен заново получить решение, иначе он постоянно спорит с уже новым владельцем. Для аудита сохраните, какой кандидат был принят и почему другой отклонён.

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

После успешного исправления сравните контрольные итоги принятой партиции с прежней версией. Это помогает подтвердить конкретную цель backfill: исправление суммы, восстановление пропущенных строк или новое правило расчёта. Сам факт смены владельца не доказывает, что исторические данные стали правильными.

В курсе есть оркестрация зависимостей, расписания и backfill через Airflow, обработка Spark и итоговый Kafka → Spark → Lakehouse pipeline. Следующая задача — встроить согласованную публикацию партиций в этот маршрут и проверить восстановление после сбоя. Откройте программу платного курса «Data Engineering Fast‑Track: Kafka → Airflow → Spark».

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