多次调用 Airflow Dag 失败回调答案

【问题标题】：Airflow Dag failure callback being called multiple times多次调用 Airflow Dag 失败回调
【发布时间】：2020-09-01 15:49:00
【问题描述】：

on_failure_callback 功能被多次调用时遇到问题。

目前，我的主要Dag分为1个PythonOperator和4个SubDag，每个SubDag包含多个PythonOperator。

t_python_1 >> t_subdag_2 >> t_subdag_3 >> t_subdag_4 >> t_subdag_5

__t_subdag_2__:
t_sub2_python_1 >> t_sub2_python_2 t_sub2_python_3

...

在设置过程中，在 Dag 的 default_args 中，我添加了一个 on_failure_callback 选项。对于每个 SubDag，我将其设置为 None。

参数主Dag:

DEFAULT_ARGS_MAINDAG = {
    "owner"               : "airflow",
    "depends_on_past"     : False,
    "start_date"          : days_ago(2),
    "retries"             : 2,
    "retry_delay"         : timedelta(seconds=30),
    "on_failure_callback" : custom_failure_handle,
}

参数SubDag:

DEFAULT_ARGS_SUBDAG = {
    "owner"               : "airflow",
    "depends_on_past"     : False,
    "start_date"          : days_ago(2),
    "retries"             : 2,
    "retry_delay"         : timedelta(seconds=30),
    "on_failure_callback" : None,
}

但是当我运行 Dag 并检查任务时，似乎每个 subdag 的任务都将 on_failure_callback 设置为 custom_failure_handle 而不是 None。所以在执行过程中，如果 SubDag 任务失败，on_failure_callback 会被调用两次：在主 Dag 和 SubDag 中。

这是否可以为父 Dag 和 SubDag 单独配置以确保失败回调不会被调用两次？

【问题讨论】：

这可能是因为您配置了"retries": 2；所以每次尝试失败都会调用on_failure_callback（尽管从技术上讲，它应该被调用三次：原始运行失败 + 2 次重试尝试失败）
@y2k-shubham 不是应该在所有重试失败后才调用吗？
好吧，我的错；我把它和on_retry_callback 混在一起了。 Zat42 是的，确实on_failure_callback 是supposed to be invoked，仅当任务最终标记为FAILED 时（不是间歇性故障，因为当时任务的状态是UP_FOR_RETRY）

标签： python airflow

【解决方案1】：

所以问题不在于Airflow 方面，而在于我实现一个为我的SubDag 创建default_args 的服务的方式。基本上SubDag default_args 总是被主要的Dag 参数覆盖。

您当然可以为每个 SubDag 传递不同的配置，方法与为常规 Dag 传递的方式相同。

【讨论】：