如何在Airflow中为任务配置execution_timeout及失败邮件通知

核心原理 #

给任务设置execution_timeout参数后，任务实际运行时长超过阈值时，Airflow会主动终止任务进程，抛出AirflowTaskTimeout异常将任务标记为失败，只要绑定任务失败的邮件通知逻辑，就能实现超时触发告警。

第一步：配置Airflow SMTP邮件服务 #

先在Airflow配置文件airflow.cfg中修改SMTP参数，确保邮件能正常发送：

[email]
email_backend = airflow.utils.email.send_email_smtp

[smtp]
smtp_host = 你的SMTP服务地址
smtp_starttls = True
smtp_ssl = False
smtp_user = 发件邮箱账号
smtp_password = 发件邮箱授权码/密码
smtp_port = SMTP服务端口
smtp_mail_from = 发件人显示邮箱

第二步：配置任务超时与失败通知 #

有两种常用配置方式，可按需选择：

方式1：使用Airflow默认邮件模板（配置简单） #

不需要自定义回调函数，直接在参数中开启失败邮件、配置接收邮箱和超时时间即可。如果要给DAG下所有任务统一生效，直接写在default_args里，单个任务特殊配置可以在任务参数中覆盖：

from datetime import datetime, timedelta
from airflow import DAG
from airflow.operators.bash import BashOperator

default_args = {
    'owner': 'airflow',
    'start_date': datetime(2024, 1, 1),
    # 配置告警接收邮箱，支持多个
    'email': ['your-alert-email@example.com'],
    # 开启任务失败发邮件（超时属于失败场景，会触发该逻辑）
    'email_on_failure': True,
    # 默认全局任务超时时间，单任务可单独覆盖
    'execution_timeout': timedelta(seconds=300)
}

with DAG(
    dag_id='timeout_email_demo',
    default_args=default_args,
    schedule_interval=None,
    catchup=False
) as dag:
    t1 = BashOperator(
        task_id='sleep_for_10s',
        bash_command='sleep 10s',
        # 单独给该任务设置5秒超时，覆盖全局默认配置
        execution_timeout=timedelta(seconds=5)
    )

方式2：自定义邮件内容（灵活度高） #

如果需要自定义告警邮件的内容、格式，可以通过on_failure_callback绑定自定义的邮件发送函数，超时触发失败时会自动执行该函数：

from datetime import datetime, timedelta
from airflow import DAG
from airflow.operators.bash import BashOperator
from airflow.utils.email import send_email

# 自定义失败告警回调
def timeout_failure_alert(context):
    ti = context.get('task_instance')
    # 组装邮件内容
    content = f"""
    <h3>Airflow任务执行失败告警</h3>
    <ul>
        <li>DAG ID: {ti.dag_id}</li>
        <li>Task ID: {ti.task_id}</li>
        <li>执行时间: {context.get('execution_date')}</li>
        <li>失败原因: {context.get('exception')}</li>
        <li>日志跳转: {ti.log_url}</li>
    </ul>
    """
    send_email(
        to=['your-alert-email@example.com'],
        subject=f"[Airflow告警] 任务{ti.dag_id}.{ti.task_id}执行失败",
        html_content=content
    )

default_args = {
    'owner': 'airflow',
    'start_date': datetime(2024, 1, 1),
    'execution_timeout': timedelta(seconds=300),
    # 绑定失败回调，所有任务失败（含超时）都会触发
    'on_failure_callback': timeout_failure_alert
}

with DAG(
    dag_id='custom_alert_demo',
    default_args=default_args,
    schedule_interval=None,
    catchup=False
) as dag:
    t1 = BashOperator(
        task_id='sleep_for_10s',
        bash_command='sleep 10s',
        execution_timeout=timedelta(seconds=5)
    )

注意事项 #

• execution_timeout是硬超时阈值，超时后Airflow会直接杀掉任务运行进程标记失败；不要和SLA配置混淆，SLA超时只会发送错过通知，不会终止任务运行。
• 配置在default_args中的参数会自动应用到DAG下所有任务，不需要给每个任务单独重复配置，单任务有特殊需求直接在任务初始化参数中重写对应值即可。
• 配置SMTP后建议先测试邮件发送通路是否正常，避免告警收不到的情况。

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