Apache Airflow：将任务延迟一段时间

Question

我试图在 DAG 内的父任务 5 分钟后执行任务。

DAG：任务 1 ----> 等待 5 分钟 ----> 任务 2

如何在 Apache Airflow 中实现这一点？提前致谢。

Answer 1

上述行为可以通过引入一项任务来实现，该任务强制您Task 1和Task 2

这可以使用 PythonOperator

import time
from airflow.operators.python_operator import PythonOperator

delay_python_task: PythonOperator = PythonOperator(task_id="delay_python_task",
                                                   dag=my_dag,
                                                   python_callable=lambda: time.sleep(300))

task_1 >> delay_python_task >> task_2

或使用BashOperator以及

from airflow.operators.bash_operator import BashOperator
delay_bash_task: BashOperator = BashOperator(task_id="delay_bash_task",
                                             dag=my_dag,
                                             bash_command="sleep 5m")
task_1 >> delay_bash_task >> task_2

注意：给定的代码片段未经测试

参考

• example_python_operator.py
• example_bash_operator.py

更新-1

以下是引入延迟的其他一些方法

• on_success_callback/ on_failure_callback：根据是否Task 2应该在 成功或失败时运行Task 1，您可以传入以下lambda: time.sleep(300)任一参数Task 1
• pre_execute()/ post_execute(): 调用time.sleep(300)in Task 1'spost_execute()或Task 2'spre_execute()也有同样的效果。当然，这将涉及为您的tasks（1 或 2）修改代码，因此最好避免它

我个人更喜欢额外的task方法，因为它使事情更加明确并且不会错误地夸大您Task 1或Task 2