Надёжные DAG: обработка ошибок и повторные попытки

Построение пайплайнов данных с Airflow

Volker Janz

Senior Developer Advocate at Astronomer

Почему задачи завершаются с ошибкой

Проблемы с подключением задач

Таймауты API и превышение лимитов запросов
Кратковременные сбои сети и задержки DNS
Конкуренция нескольких задач за один ресурс

Повторные попытки

@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():
        ...

Повторы + выдержка справляются с временными сбоями
Колбэки уведомляют нужных людей
Автопауза устраняет лишний шум

Давайте потренируемся!

Построение пайплайнов данных с Airflow