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

问题描述 投票:0回答:1

在我的

Airflow DAG
我有4个
tasks

task_1 >> [task_2,task_3]>> task_4

task_4
仅在
task_2
task_3

成功运行后运行

我如何设置一个条件,例如:

如果

task_2
失败,请在2分钟后重试
task_2
,并在第5次尝试后停止重试

这是我的代码:

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

airflow-scheduler airflow
1个回答
17
投票

每个操作员都支持

retry_delay
retries
- 气流文档

retries (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
,而是设置固定的绝对日期。

© www.soinside.com 2019 - 2024. All rights reserved.