Airflow中能否获取任务前一次重试的错误信息？

当然可以在任务第2次运行时提取第一次尝试的异常，核心思路是在重试过程中保留每次失败的异常记录，下面分常见场景给出具体实现：

基于重试框架（以Python tenacity为例） #

如果使用tenacity这类成熟的重试库，它提供了retry_state参数来访问重试上下文，其中包含所有历史尝试的异常信息：

from tenacity import retry, stop_after_attempt, retry_if_exception_type

# 定义你的特定异常类型
class TargetException(Exception):
    pass

def handle_target_exception(exc):
    # 针对特定异常的自定义逻辑
    print(f"触发特定异常处理: {str(exc)}")

@retry(stop=stop_after_attempt(3), retry=retry_if_exception_type(TargetException))
def business_task():
    # 模拟任务抛出特定异常
    raise TargetException("初始执行失败")

# 绑定重试回调，在指定重试次数时处理历史异常
@business_task.retry_callback
def on_retry(retry_state):
    # 当前是第2次尝试时，提取第一次的异常
    if retry_state.attempt_number == 2:
        first_failure_exc = retry_state.previous_attempts[0].exception()
        if isinstance(first_failure_exc, TargetException):
            handle_target_exception(first_failure_exc)

# 执行任务
try:
    business_task()
except TargetException:
    # 最终重试失败后的处理（可选）
    pass

自定义重试逻辑 #

如果是自己实现重试循环，只需要在每次失败时保存异常实例即可：

def business_task():
    raise TargetException("初始执行失败")

max_retry_times = 3
failure_records = []

for attempt in range(1, max_retry_times + 1):
    try:
        business_task()
        break  # 执行成功，跳出循环
    except Exception as exc:
        failure_records.append(exc)
        # 第2次尝试失败后，检查第一次的异常
        if attempt == 2:
            first_exc = failure_records[0]
            if isinstance(first_exc, TargetException):
                handle_target_exception(first_exc)
        # 达到最大重试次数，抛出最后一次异常
        if attempt == max_retry_times:
            raise exc

核心注意点 #

• 无论用框架还是自定义实现，保存每次失败的异常实例是关键，后续重试步骤才能按需提取
• 用isinstance()判断异常类型，比直接匹配异常消息更可靠
• 注意区分当前尝试的异常和历史尝试的异常，避免逻辑混淆

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