Airflow backfill дублирует строки: загрузка по data interval

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

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

Дата запуска и дата данных — разные величины

При backfill код выполняется сегодня, но обрабатывает прошлые даты. Поэтому datetime.now() нельзя использовать для выбора входного дня. Берите границы из data_interval_start и data_interval_end текущего запуска. Для обычного дневного расписания это начало и конец обрабатываемого дня в часовом поясе расписания.

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

Проверьте повтор и сбой без Airflow

Сохраните следующий код как daily_load.py и запустите Python. Он работает в SQLite в памяти. Все даты записаны в одинаковом ISO-формате UTC, чтобы строковое сравнение совпадало со сравнением времени. Для настоящего хранилища используйте тип timestamp и параметризованные границы.

import sqlite3

SOURCE_ROWS = [
    ("A", "2026-10-01T12:00:00+00:00", 100),
    ("B", "2026-10-02T00:00:00+00:00", 200),
]

def replace_interval(con, start, end, source_rows, fail=False):
    # Все timestamps: фиксированный ISO-формат, UTC, одинаковая точность.
    if start >= end:
        raise ValueError("Empty or reversed interval")
    rows = [r for r in source_rows if start <= r[1] < end]
    if len({r[0] for r in rows}) != len(rows):
        raise ValueError("Duplicate event id in source snapshot")
    with con:
        con.execute("DELETE FROM events WHERE event_time >= ? AND event_time < ?",
                    (start, end))
        if fail:
            raise RuntimeError("Synthetic failure after delete")
        con.executemany("INSERT INTO events VALUES (?, ?, ?)", rows)

if __name__ == "__main__":
    con = sqlite3.connect(":memory:")
    con.execute("CREATE TABLE events (id TEXT PRIMARY KEY, event_time TEXT, amount INTEGER)")
    start = "2026-10-01T00:00:00+00:00"
    end = "2026-10-02T00:00:00+00:00"
    snapshot = lambda: con.execute("SELECT * FROM events ORDER BY id").fetchall()
    replace_interval(con, start, end, SOURCE_ROWS)
    first = snapshot()
    replace_interval(con, start, end, SOURCE_ROWS)
    assert snapshot() == first == [SOURCE_ROWS[0]]
    try:
        replace_interval(con, start, end, [], fail=True)
    except RuntimeError:
        pass
    assert snapshot() == first
    replace_interval(con, end, "2026-10-03T00:00:00+00:00", SOURCE_ROWS)
    assert snapshot() == SOURCE_ROWS
    replace_interval(con, start, end, [])
    assert snapshot() == [SOURCE_ROWS[1]]
    print("retry, rollback, boundary and empty snapshot passed")

Вывод подтверждает четыре свойства. Второй вызов с тем же снимком оставляет прежние строки. Искусственная ошибка после DELETE откатывает удаление. Событие B на правой границе попадает только в следующий интервал. Пустой подтверждённый снимок удаляет старый результат за свой интервал, сохраняя соседний.

Последний случай полезен при исправлении источника: обычный upsert добавляет и обновляет строки, но не обязательно удаляет события, исчезнувшие из нового снимка. Полная замена подходит, когда источник действительно отдаёт весь интервал. Если пустой ответ означает недоступность API, отличите его от корректного пустого снимка до начала транзакции. Иначе сбой чтения превратится в успешное удаление витрины.

Передайте интервал из DAG

Для Airflow 3.x подключение выглядит так. Файл daily_load.py должен быть доступен для импорта. Этот локальный учебный DAG использует SQLite-файл в существующем каталоге /opt/airflow/data; все исполнения должны видеть один и тот же файл. В распределённой установке замените его общим целевым хранилищем и соответствующим Hook.

from datetime import datetime, timezone
import sqlite3
from airflow.sdk import dag, task, get_current_context
from daily_load import replace_interval, SOURCE_ROWS

@dag(schedule="@daily", start_date=datetime(2026, 10, 1, tzinfo=timezone.utc),
     catchup=False, max_active_runs=1)
def daily_events():
    @task(retries=2)
    def load():
        context = get_current_context()
        start = context["data_interval_start"].in_timezone("UTC").isoformat()
        end = context["data_interval_end"].in_timezone("UTC").isoformat()
        with sqlite3.connect("/opt/airflow/data/demo.db") as con:
            con.execute("CREATE TABLE IF NOT EXISTS events (id TEXT PRIMARY KEY, event_time TEXT, amount INTEGER)")
            replace_interval(con, start, end, SOURCE_ROWS)
    load()

daily_events()

Константа SOURCE_ROWS здесь задаёт синтетический снимок для проверки связи DAG с загрузчиком. В рабочей задаче сначала прочитайте полный снимок нужного интервала из источника, проверьте его готовность и только затем вызывайте replace_interval. Время создания запуска не должно подменять границы данных.

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

Учитывайте параллельные пересчёты

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

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

В платном курсе Data Engineering Fast-Track: Kafka → Airflow → Spark вы разберёте расписания и надёжность Airflow, обработку данных в Spark и хранение в lakehouse, затем соедините компоненты в end-to-end конвейер. В материалах курса предусмотрен сценарий backfill. Откройте программу курса, чтобы отработать повторную обработку вместе с источником и целевым хранилищем.