如何在气流DAG中设置数字作为重试条件？

问题描述

在我的Airflow DAG中，我有4个tasks

task_1 >> [task_2,task_3]>> task_4

task_4仅在成功运行task_2和task_3

之后运行

如何设置条件：

如果task_2失败，请在2分钟后重试task_2，并在第五次尝试后停止重试

这是我的代码：

from airflow.models import DAG
from airflow.utils.dates import days_ago
from airflow.operators.python_operator import Pythonoperator

args={
    'owner' : 'Anti','start_date':days_ago(1)# 1 means yesterday
}

dag = DAG(dag_id='my_sample_dag',default_args=args,schedule_interval='15 * * * *')

def func1(**context):
    print("ran task 1")

def func2(**context):
    print("ran task 2")

def func3(**context):
    print("ran task 3")

def func4(**context):
    print("ran task 4")

with dag:
    task_1=Pythonoperator(
        task_id='task1',python_callable=func1,provide_context=True,)
    task_2=Pythonoperator(
        task_id='task2',python_callable=func2,provide_context=True 
    )
    task_3=Pythonoperator(
        task_id='task3',python_callable=func3,provide_context=True 
    )
    task_4=Pythonoperator(
        task_id='task4',python_callable=func4,provide_context=True 
    )

task_1 >> [task_2,task_3]>> task_4 # t2,t3 runs parallel right after t1 has ran

解决方法

每个运算符都支持retry_delay和retries-Airflow documention。

重试（int）–在执行之前应重试的次数 任务失败

retry_delay（datetime.timedelta）–重试之间的延迟

如果要将其应用于所有任务，则只需编辑args字典：

args={
    'owner' : 'Anti','retries': 5,'retry_delay': timedelta(minutes=2),'start_date':days_ago(1)# 1 means yesterday
}

如果您只想将其应用于task_2，则可以将其直接传递给PythonOperator-在这种情况下，其他任务将使用默认设置。

对您的参数发表评论，不建议设置动态相对值start_date，而是设置固定的绝对日期。