Airflow中如何实现满足条件后重试任务？跨DAG依赖场景

针对你描述的场景，有几种高效方案可以避免Task B的无效重试，仅在DAG A完成运行后再执行/重试Task B：

方案1：使用ExternalTaskSensor监听DAG A的成功状态 #

把Task B的执行逻辑拆分为等待DAG A完成和执行消费逻辑两个步骤，通过ExternalTaskSensor监听DAG A的最新成功运行，只有当DAG A的Task A成功完成后，才会触发Task B的执行。这种方式从根源上避免了Task B因数据未生成而失败的情况，无需依赖重试机制。

示例代码：

from airflow import DAG
from airflow.sensors.external_task import ExternalTaskSensor
from airflow.operators.python import PythonOperator
from datetime import datetime, timedelta

def task_b_consume_data():
    # 这里写Task B的消费数据逻辑
    pass

default_args = {
    'owner': 'airflow',
    'depends_on_past': False,
    'start_date': datetime(2024, 1, 1),
    'retries': 0,  # 无需设置重试，传感器会等待到条件满足
}

with DAG(
    'dag_b',
    default_args=default_args,
    schedule_interval='@hourly',
    catchup=False
) as dag_b:
    # 监听DAG A的Task A最新成功运行
    wait_for_dag_a = ExternalTaskSensor(
        task_id='wait_for_dag_a',
        external_dag_id='dag_a',
        external_task_id='task_a',
        mode='reschedule',  # 释放worker资源，每隔一段时间再检查
        poke_interval=300,  # 每5分钟检查一次DAG A的状态
        timeout=86400,  # 最多等待1天，避免无限挂起
        allowed_states=['success'],
        failed_states=['failed', 'skipped'],
    )

    task_b = PythonOperator(
        task_id='task_b',
        python_callable=task_b_consume_data,
    )

    wait_for_dag_a >> task_b

优点：逻辑直观，完全避免无效重试，资源消耗可控；缺点：如果DAG A长时间不运行，传感器会保留调度记录（但reschedule模式会释放worker资源）。

方案2：通过DAG A的成功回调触发DAG B的重试 #

在DAG A的Task A完成后，主动触发DAG B中对应小时的失败实例进行重试。这种方式适合DAG B已经因数据缺失失败的场景，无需让DAG B一直等待，而是等DAG A生成数据后再唤醒重试。

示例代码：

from airflow import DAG
from airflow.operators.python import PythonOperator
from airflow.api.client.local_client import Client
from datetime import datetime, timedelta

def task_a_generate_data():
    # Task A的生成数据逻辑
    pass

def trigger_dag_b_retry(context):
    """DAG A成功后，触发DAG B对应小时的失败实例重试"""
    client = Client(None, None)
    # 获取当前执行小时对应的DAG B失败实例
    target_exec_date = context['execution_date'].replace(minute=0, second=0, microsecond=0)
    dag_b_runs = client.get_dag_runs(
        dag_id='dag_b',
        state='failed',
        execution_date_gte=target_exec_date,
        execution_date_lte=target_exec_date + timedelta(hours=1)
    )
    # 触发每个失败实例的重试
    for run in dag_b_runs:
        client.trigger_dag_run(
            dag_id='dag_b',
            run_id=run.run_id,
            conf={'retry_triggered_by_dag_a': True},
            replace_microseconds=False
        )

default_args = {
    'owner': 'airflow',
    'start_date': datetime(2024, 1, 1),
}

with DAG(
    'dag_a',
    default_args=default_args,
    schedule_interval=None,  # 手动触发
    catchup=False
) as dag_a:
    task_a = PythonOperator(
        task_id='task_a',
        python_callable=task_a_generate_data,
        on_success_callback=trigger_dag_b_retry,  # 成功后触发DAG B重试
    )

优点：DAG B无需等待，失败后直接结束，等DAG A运行后再重试，资源利用率更高；缺点：需要确保DAG B的失败实例和DAG A的执行时间对应准确，避免触发错误的实例。

方案3：自定义Task B的重试判断逻辑 #

在Task B的代码中，先检查DAG A是否已经运行并生成了对应小时的数据（比如检查数据文件是否存在，或者通过Airflow的XCom/DAG Run记录判断），如果数据未生成，就抛出特定异常，并设置合理的重试间隔。不过这种方式不如前两种可靠，仅作为备选。

示例代码片段：

def task_b_consume_data():
    # 自定义检查逻辑：判断DAG A是否生成了对应数据
    data_exists = check_data_exists()
    if not data_exists:
        # 抛出异常触发重试
        raise ValueError("Data not generated by DAG A yet")

task_b = PythonOperator(
    task_id='task_b',
    python_callable=task_b_consume_data,
    retries=24,  # 最多重试24次（覆盖一整天）
    retry_delay=timedelta(hours=1),
    retry_exponential_backoff=True,
)

优点：无需修改DAG结构，仅在Task B内部处理；缺点：仍可能产生少量无效重试，且重试次数和间隔需要提前配置。

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