如何为Airflow不同任务配置专属失败通知收件人？

我有一个包含多个任务的Airflow DAG，每个任务对应不同的负责人。当任务失败时，我希望通知对应的负责人。我尝试在创建DAG时传入default_args，其中配置了email参数，但无论哪个任务失败，都会发送邮件给单个用户；虽可添加多个用户，但无法实现专属通知。请问能否在任务级别配置邮箱？

尝试的代码如下：

default_args = {
    "retries": 1,
    "email_on_failure": True,
    "email": ['abc@gmail.com'],
    "get_logs": True,
    "concurrency": 1,
    "max_active_runs": 1,
    "startup_timeout_seconds": 3600,
    "labels": {'managedby': 'airflow'},
    "on_failure_callback": on_failure_callback_slack
}

dag = DAG(dag_id="abc", default_args=default_args, schedule_interval="30 20 * * *", start_date="2023-07-05")

解决方案 #

当然可以在任务级别配置专属邮箱，只需在定义每个任务时，单独指定email和email_on_failure参数，覆盖default_args中的全局配置即可。

示例代码如下：

default_args = {
    "retries": 1,
    "email_on_failure": True,  # 全局默认开启失败邮件通知
    "get_logs": True,
    "concurrency": 1,
    "max_active_runs": 1,
    "startup_timeout_seconds": 3600,
    "labels": {'managedby': 'airflow'},
    "on_failure_callback": on_failure_callback_slack
}

dag = DAG(dag_id="abc", default_args=default_args, schedule_interval="30 20 * * *", start_date="2023-07-05")

# 任务1：负责人A
task_a = BashOperator(
    task_id="task_a",
    bash_command="echo 'Task A running...'",
    email=['a_user@gmail.com'],  # 任务专属邮箱
    dag=dag
)

# 任务2：负责人B
task_b = PythonOperator(
    task_id="task_b",
    python_callable=some_python_function,
    email=['b_user@gmail.com'],  # 任务专属邮箱
    dag=dag
)

额外说明 #

• 如果某个任务不需要失败通知，可单独设置email_on_failure=False覆盖全局配置
• 每个任务的email参数支持传入列表，可同时通知多个负责人
• 这种配置方式会让任务失败时，仅发送邮件给该任务指定的专属邮箱，实现精准的负责人通知

内容的提问来源于stack exchange，提问作者monte