如何配置Airflow告警仅在最后一次重试失败时触发？

可以实现，核心是通过Airflow任务实例（TaskInstance）的重试相关属性，在回调函数里判断当前是否是最后一次失败尝试，再执行告警逻辑。

修改回调函数逻辑 #

在原有的on_attempted_download_failure_callback函数中，增加对当前重试次数的判断，只有当当前尝试是最后一次失败时，才执行告警逻辑：

def on_attempted_download_failure_callback(context):
    """
    The callback will issue slack messages and phone calls if current time
    is beyond a tolerance threshold while downloads are still failing.
    """
    ti = context['ti']
    # 判断当前是否是最后一次重试失败
    if ti.is_last_try():
        tolerance = get_downloads_tolerance()

        # The DAG starts running one schedule interval after its execution date
        # so the timestamp to compare to is the next execution date if we want a
        # threshold of N hours since the DAG started running.
        now_utc = pendulum.now("UTC")
        if now_utc - context['next_execution_date'] > timedelta(hours=tolerance):
            issue_alerts(context)

你也可以直接通过try_number和max_tries的对比来判断（效果和is_last_try()完全一致）：

def on_attempted_download_failure_callback(context):
    """
    The callback will issue slack messages and phone calls if current time
    is beyond a tolerance threshold while downloads are still failing.
    """
    ti = context['ti']
    # retries=5时，max_tries=6（1次初始尝试+5次重试），当try_number等于max_tries时就是最后一次失败
    if ti.try_number == ti.max_tries:
        tolerance = get_downloads_tolerance()

        now_utc = pendulum.now("UTC")
        if now_utc - context['next_execution_date'] > timedelta(hours=tolerance):
            issue_alerts(context)

原理说明 #

• Airflow中，任务实例的try_number从1开始计数（第一次尝试是1，第一次重试是2，以此类推）。
• max_tries的值等于你设置的retries参数加1（比如retries=5时，max_tries=6，对应1次初始尝试+5次重试）。
• is_last_try()是Airflow封装的判断方法，内部逻辑就是对比try_number和max_tries，用起来更直观。

修改后，只有当任务耗尽所有重试次数最终失败时，才会触发告警逻辑，之前的重试失败不会触发。

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