如何捕获Airflow任务错误日志并通过邮件发送异常详情？

要在Airflow任务失败邮件中包含UI里显示的完整异常栈信息，不能用task_instance.log.exception（这是日志记录方法，仅返回方法绑定信息），正确做法是从任务上下文的exception对象中提取，并用traceback模块格式化完整栈内容。

修改步骤 #

• 导入traceback模块，用于格式化异常栈文本
• 从context中获取exception对象，生成完整异常栈信息
• 将异常详情插入到邮件内容模板中

修改后的完整代码 #

import traceback

def failure_callback(context):
    ti = context.get('task_instance')
    # 获取并格式化完整异常栈
    exception = context.get('exception')
    error_trace = traceback.format_exc() if exception else "未捕获到异常信息"
    
    failure_msg = """
            :red_circle: 任务执行失败
            *DAG ID*: {dag}  
            *任务ID*: {task}
            *执行时间*: {exec_date}
            *日志链接*: {log_url}
            *异常详情*:
            ```
            {error_trace}
            ```
            """.format(
        dag=ti.dag_id,
        task=ti.task_id,
        exec_date=context.get('execution_date'),
        log_url=ti.log_url,
        error_trace=error_trace
    )
    # 此处添加发送邮件的逻辑（如调用SMTP服务发送failure_msg）

关键说明 #

• context.get('exception')可直接获取任务抛出的异常对象
• traceback.format_exc()会生成与Airflow UI日志完全一致的异常栈文本，包含Traceback调用链和错误详情
• 增加异常为空的兜底处理，避免邮件内容缺失

测试效果验证 #

使用你提供的测试异常代码执行任务后，邮件的异常详情部分会显示与UI一致的内容：

[2023-01-04, 09:05:07 UTC] {taskinstance.py:1909} ERROR - Task failed with exception
Traceback (most recent call last):
  File "/opt/bitnami/airflow/dags/git_airflow-dags/scan_table.py", line 37, in scan_table
    raise ValueError('File not parsed completely/correctly')
ValueError: File not parsed completely/correctly

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