在我的
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
每个操作员都支持
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
,而是设置固定的绝对日期。