针对你遇到的Airflow Python组件没触发邮件告警的问题，我整理了几个无需重启Airflow服务、运行时即可生效的解决方案，帮你快速配置告警：

1. 直接在Python任务中配置告警参数 #

这是最直接的方案——给你的PythonOperator显式添加邮件告警相关参数，只要DAG被Airflow加载（比如代码更新后自动刷新或手动触发刷新）就会生效，完全不需要修改全局配置：

from airflow.operators.python import PythonOperator
from airflow import DAG
from datetime import datetime

def my_python_task():
    # 这里写你的Python任务逻辑，模拟失败场景
    raise ValueError("模拟Python任务执行失败")

with DAG(
    dag_id="python_task_alert_demo",
    start_date=datetime(2024, 1, 1),
    schedule_interval=None,
) as dag:
    python_alert_task = PythonOperator(
        task_id="python_failure_alert",
        python_callable=my_python_task,
        # 指定告警邮箱
        email=["your-alert-email@example.com", "team-alert@example.com"],
        # 任务失败时发送邮件
        email_on_failure=True,
        # 任务重试时发送邮件（可选）
        email_on_retry=True,
        # 重试次数（可选）
        retries=1
    )

2. 用Airflow Variables动态管理告警邮箱 #

如果不想把邮箱地址硬编码在DAG里，想随时在UI里修改告警收件人，可以用Airflow的Variables功能，运行时修改立即生效：

1. 先打开Airflow UI，进入「Admin -> Variables」，创建一个变量：

   • Key: PYTHON_TASK_ALERT_EMAILS
   • Value: alert1@example.com,alert2@example.com（逗号分隔多个邮箱）

2. 在DAG代码中读取这个变量，配置到任务里：

from airflow.operators.python import PythonOperator
from airflow.models import Variable
from airflow import DAG
from datetime import datetime

def my_python_task():
    # 你的任务逻辑
    pass

# 从Airflow变量中读取告警邮箱
alert_emails = Variable.get(
    "PYTHON_TASK_ALERT_EMAILS",
    default_var="default-alert@example.com"
).split(",")

with DAG(
    dag_id="dynamic_email_alert_demo",
    start_date=datetime(2024, 1, 1),
    schedule_interval=None,
) as dag:
    python_task = PythonOperator(
        task_id="dynamic_alert_task",
        python_callable=my_python_task,
        email=alert_emails,
        email_on_failure=True,
        dag=dag
    )

之后要修改收件人，直接在UI里更新变量值就行，不用改DAG代码。

3. 自定义失败回调函数实现灵活告警 #

如果默认的告警邮件内容太简单，想添加任务日志链接、执行上下文等信息，可以自定义失败回调函数，用Airflow自带的send_email工具发送定制化邮件：

from airflow.operators.python import PythonOperator
from airflow.utils.email import send_email
from airflow import DAG
from datetime import datetime

def custom_failure_alert(context):
    # 从上下文获取任务相关信息
    task_instance = context["task_instance"]
    dag_id = task_instance.dag_id
    task_id = task_instance.task_id
    execution_date = task_instance.execution_date
    log_url = task_instance.log_url

    # 定制邮件内容
    subject = f"⚠️ Airflow任务失败: {dag_id}.{task_id}"
    html_content = f"""
    <h3>任务失败详情</h3>
    <ul>
        <li>DAG ID: {dag_id}</li>
        <li>任务ID: {task_id}</li>
        <li>执行时间: {execution_date}</li>
        <li>查看日志: <a href="{log_url}">点击跳转</a></li>
    </ul>
    """
    # 发送邮件
    send_email(to=["your-alert@example.com"], subject=subject, html_content=html_content)

def my_python_task():
    # 模拟任务失败
    raise Exception("Python任务执行出错")

with DAG(
    dag_id="custom_alert_demo",
    start_date=datetime(2024, 1, 1),
    schedule_interval=None,
) as dag:
    python_task = PythonOperator(
        task_id="custom_alert_task",
        python_callable=my_python_task,
        # 指定自定义失败回调
        on_failure_callback=custom_failure_alert,
        dag=dag
    )

这个方案不仅能实现邮件告警，还能根据需求扩展其他通知方式（比如企业微信、Slack），灵活性很高。

额外检查提示 #

虽然你的EMR任务能正常收到邮件，但还是建议确认下Airflow的全局SMTP配置是否正确（比如smtp_host、smtp_port、smtp_user等）。如果需要运行时调整这些配置，可以通过Airflow UI的「Admin -> Configurations」修改，多数邮件相关配置支持热加载，无需重启服务。

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