Надійні Dag з обробкою збоїв і повторними спробами

Створення конвеєрів даних з Airflow

Volker Janz

Senior Developer Advocate at Astronomer

Чому завдання падають

Проблеми зі з'єднанням завдань

Тайм-аути API та ліміти запитів
Короткі збої мережі й затримки розв'язання доменних імен
Кілька завдань змагаються за один ресурс

Повторні спроби

@task(
    retries=3,
    retry_delay=timedelta(minutes=2),
)
def fetch_weather():
    response = requests.get("https://api.weather.com/forecast")
    response.raise_for_status()
    return response.json()

Завдання пробує ще до 3 разів після збою
Очікує 2 хвилини між спробами

Експоненційне відступання

@task(
    retries=3,
    retry_delay=timedelta(minutes=2),
    retry_exponential_backoff=2.0,
)
def fetch_weather():
    ...

Експоненційне відступання

Затримка подвоюється після кожної спроби
Дає зовнішнім сервісам час відновитись

on_failure_callback

def alert_on_failure(context):
    dag_id = context["dag"].dag_id
    task_id = context["ti"].task_id
    print(f"ALERT: {task_id} in {dag_id} failed!")

@task(on_failure_callback=alert_on_failure)
def fetch_weather():
    ...

Словник context: dag, ti, exception, URL журналу
Маршрутизуйте в Slack, PagerDuty або інший інструмент сповіщень

Зворотні виклики на рівні Dag і завдання

Рівень Dag (універсальний)

@dag(
    on_failure_callback=alert_team,
)
def my_pipeline():
    ...

Рівень завдання (точковий)

@task(
    on_failure_callback=page_oncall,
)
def critical_step():
    ...

Рівень завдання перекриває рівень Dag
Використовуйте обидва для багаторівневих сповіщень

max_consecutive_failed_dag_runs

@dag(max_consecutive_failed_dag_runs=3)
def monitoring_pipeline():
    ...

Невдалі запуски Dag з подальшою автопаузою

Запобігає втомі від сповіщень через постійні збої
Dag відновиться після ручного зняття з паузи

Об'єднаймо все разом

@dag(
    max_consecutive_failed_dag_runs=3,
    on_failure_callback=alert_team,
)
def weather_pipeline():

    @task(
        retries=3,
        retry_delay=timedelta(minutes=2),
        retry_exponential_backoff=True,
        on_failure_callback=page_oncall,
    )
    def fetch_weather():
        ...

Повторні спроби + backoff обробляють тимчасові збої
Зворотні виклики сповіщають потрібних людей
Автопауза зменшує «шум»

Давайте потренуємось!

Створення конвеєрів даних з Airflow