Airflow能否让下游任务在上游任务失败时发送错误邮件通知？

问题根因 #

email_on_failure=True仅在任务自身执行出错进入failed状态时触发告警，t5执行失败后，下游t3会直接进入upstream_failed状态，不会实际运行，因此不会触发自身的失败告警逻辑。

解决方案 #

方案1：DAG层级全局失败告警（最简便）

给整个DAG添加失败回调，只要DAG内任意任务失败就触发邮件通知，无需单独给下游任务配置：

1. 首先定义告警回调函数

from airflow.utils.email import send_email

def dag_failure_alert(context):
    dag_id = context['dag_run'].dag_id
    failed_task = context['task_instance'].task_id
    execution_date = context['execution_date']
    log_url = context['task_instance'].log_url

    subject = f"Airflow DAG失败告警：{dag_id}"
    html_content = f"""
    <p>DAG名称：{dag_id}</p>
    <p>失败任务：{failed_task}</p>
    <p>执行时间：{execution_date}</p>
    <p>日志地址：{log_url}</p>
    """
    send_email(to="你的告警接收邮箱@xxx.com", subject=subject, html_content=html_content)

2. 给DAG添加回调配置

dag_alert_task_callback = DAG(
    'dag_alert_task_callback',
    default_args=default_args,
    schedule_interval=None,
    on_failure_callback=dag_failure_alert # 新增该行配置
)

方案2：给下游任务添加上游失败回调

如果仅需要上游失败时指定下游任务触发告警，可给t3添加上游失败回调（仅Airflow 2.x+支持该参数）：

1. 定义上游失败回调函数

from airflow.utils.email import send_email

def upstream_fail_alert(context):
    task_id = context['task_instance'].task_id
    subject = f"Airflow任务上游失败告警：{task_id}"
    html_content = f"<p>任务{task_id}的上游依赖执行失败，任务终止运行</p>"
    send_email(to="你的告警接收邮箱@xxx.com", subject=subject, html_content=html_content)

2. 修改t3的参数配置

t3 = PostgresOperator(task_id='create_aisles',
                      sql="create table if not exists dbt_raw_data.aisles (aisle_id integer, aisle varchar(100) );",
                      postgres_conn_id='dbt_postgres_instance_raw_data',
                      autocommit=True,
                      database="dbtdb",
                      dag=dag_alert_task_callback,
                      email_on_failure=True,
                      on_upstream_failure_callback=upstream_fail_alert # 新增该行配置
)

方案3：新增独立告警任务

如果需要自定义告警触发范围，可新增专门的告警任务，设置触发规则为上游任意任务失败就执行：

from airflow.operators.python import PythonOperator
from airflow.utils.trigger_rule import TriggerRule

def send_fail_alert():
    send_email(to="你的告警接收邮箱@xxx.com", subject="DAG任务失败告警", html_content="<p>DAG存在任务执行失败，请检查</p>")

# 定义告警任务
alert_task = PythonOperator(
    task_id='fail_alert',
    python_callable=send_fail_alert,
    trigger_rule=TriggerRule.ONE_FAILED,
    dag=dag_alert_task_callback
)

# 关联所有需要监控的任务到告警任务
[t2, t5, t3, t4] >> alert_task

前置检查 #

配置前请确认Airflow服务的邮件功能已经正常配置：

• airflow.cfg中smtp相关参数（smtp_host、smtp_port、smtp_user、smtp_password、smtp_mail_from）填写正确
• 已经测试过Airflow可正常发送测试邮件

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