具备失败处理与重试的健壮 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 与 Task 级回调对比

Dag 级（兜底）

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

Task 级（特定）

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

Task 级会覆盖 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 构建数据流水线