编辑“︁Airflow任务重试机制”︁

= Airflow任务重试机制 =

== 介绍 ==
'''Airflow任务重试机制'''是Apache Airflow的核心功能之一，用于处理任务执行过程中可能出现的临时性故障（如网络波动、资源不足等）。当任务失败时，Airflow可根据预设策略自动重试，而非直接标记为失败，从而提高工作流的鲁棒性。此机制特别适合对[[ETL]]、[[数据管道]]等容错性要求高的场景。

== 核心参数 ==
以下是与重试相关的关键DAG参数（定义在`default_args`或任务级别）：

<syntaxhighlight lang="python">
default_args = {
    'retries': 3,                  # 最大重试次数
    'retry_delay': timedelta(minutes=5),  # 重试间隔
    'retry_exponential_backoff': True,    # 是否启用指数退避
    'max_retry_delay': timedelta(hours=1) # 最大重试间隔
}
</syntaxhighlight>

== 重试策略详解 ==

=== 基础重试 ===
当任务失败时，Airflow会：
# 检查剩余重试次数（`retries`）
# 等待`retry_delay`指定的时间
# 重新执行任务

=== 指数退避 ===
启用`retry_exponential_backoff`后，重试间隔按公式增长：
<math>
delay = min(retry\_delay \times 2^{(retry\_attempt-1)}, max\_retry\_delay)
</math>

示例演变过程：
<mermaid>
gantt
    title 指数退避重试示例（初始间隔5分钟）
    dateFormat  HH:mm
    section 任务尝试
    第一次失败 :a1, 00:00, 1m
    第一次重试 :a2, after a1, 5m
    第二次重试 :a3, after a2, 10m
    第三次重试 :a4, after a3, 20m
</mermaid>

== 代码示例 ==

=== 基础配置 ===
<syntaxhighlight lang="python">
from datetime import datetime, timedelta
from airflow import DAG
from airflow.operators.python import PythonOperator

def unreliable_task():
    import random
    if random.random() < 0.3:  # 模拟30%失败率
        raise ValueError("Random failure")

default_args = {
    'owner': 'airflow',
    'retries': 3,
    'retry_delay': timedelta(minutes=1),
}

with DAG(
    'retry_demo',
    default_args=default_args,
    schedule_interval='@daily',
    start_date=datetime(2023, 1, 1),
) as dag:
    
    task = PythonOperator(
        task_id='unstable_task',
        python_callable=unreliable_task,
    )
</syntaxhighlight>

=== 高级配置 ===
<syntaxhighlight lang="python">
# 启用指数退避和最大延迟限制
default_args = {
    'retries': 5,
    'retry_delay': timedelta(seconds=30),
    'retry_exponential_backoff': True,
    'max_retry_delay': timedelta(minutes=10)
}
</syntaxhighlight>

== 失败原因分析 ==
Airflow记录每次重试的日志，可通过Web UI查看：
* '''Last Log'''：显示最近一次尝试的详细日志
* '''Try Index'''：指示当前是第几次重试（从1开始）

== 实际应用案例 ==

=== 案例1：API调用容错 ===
处理不稳定的第三方API时，典型配置：
* 初始重试间隔：10秒
* 最大重试次数：5
* 启用指数退避
* 最大延迟：1小时

=== 案例2：数据库连接 ===
应对数据库临时连接问题：
* 快速初始重试（5秒间隔）
* 大量重试次数（10次）
* 配合`execution_timeout`防止无限阻塞

== 最佳实践 ==
1. '''区分错误类型'''：通过`on_failure_callback`区分可重试错误与致命错误
2. '''监控重试率'''：高重试率可能表明系统设计问题
3. '''合理设置上限'''：避免无限重试消耗资源
4. '''任务幂等性'''：确保重试不会导致重复副作用

== 常见问题 ==

'''Q: 如何判断任务是否在重试状态？'''
A: 在任务中使用`context['ti'].try_number`获取当前尝试次数：
<syntaxhighlight lang="python">
def callback(context):
    print(f"Current retry attempt: {context['ti'].try_number}")
</syntaxhighlight>

'''Q: 重试会重新执行整个DAG吗？'''
A: 不会，仅重试失败的具体任务，后续任务仍按依赖关系执行。

== 进阶配置 ==
可通过自定义`retry_handler`实现更复杂的逻辑：
<syntaxhighlight lang="python">
def custom_retry_handler(context):
    exception = context['exception']
    if isinstance(exception, TimeoutError):
        return True  # 允许重试
    return False  # 其他错误不重试

default_args = {
    'retry_handler': custom_retry_handler
}
</syntaxhighlight>

== 可视化重试流程 ==
<mermaid>
stateDiagram-v2
    [*] --> Pending
    Pending --> Running: 开始执行
    Running --> Failed: 任务失败
    Failed --> Retrying: 有剩余重试次数
    Retrying --> Running: 等待重试延迟后
    Failed --> [*]: 无剩余重试次数
    Running --> Success: 执行成功
</mermaid>

[[Category:大数据框架]]
[[Category:Airflow]]
[[Category:Airflow DAG开发]]