Airflow Dag 失败回调被多次调用

Question

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

...

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

主要参数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并检查任务时，似乎每个子dag 的任务都on_failure_callback设置为custom_failure_handle而不是None。因此，在执行过程中，如果SubDag任务失败，on_failure_callback则会被调用两次：在主Dag中和在SubDag中。

是否可以对父Dag和SubDag进行单独的配置，以确保失败回调不会被调用两次？

Answer 1

所以问题不在Airflow一边，而是在我实现为我的SubDag. 基本上SubDag default_args总是被主要Dag参数覆盖。当然，您可以通过与SubDag常规Dag.