在 Airflow 中完成上一次运行 2 分钟后继续运行 DAG答案

【问题标题】：Continuously run a DAG after 2 minutes of completion of the previous run in Airflow在 Airflow 中完成上一次运行 2 分钟后继续运行 DAG
【发布时间】：2021-05-08 09:17:28
【问题描述】：

我想知道我们是否可以安排 DAG 在 Airflow 中完成同一 DAG 2 分钟后连续运行。

编辑：

我的 DAG 应该以这样的方式运行，即每次完成运行时，它都必须等待 2 分钟并重新开始运行。我不想安排我的 DAG 每 2 分钟运行一次，而是应该在同一 DAG 完成 2 分钟后立即连续运行。

【问题讨论】：

标签： google-cloud-platform airflow scheduler directed-acyclic-graphs

【解决方案1】：

您可以在一天中的任意时间安排您的 dag，并使用TriggerDagRunOperator 再次触发自身。要在触发之前等待 2 分钟，您可以简单地引入一个睡眠任务。

DAG：

任务 1 >> 任务 2 >> 任务 3 BashOperator(bash_command="sleep 120") >> 任务 4 TriggerDagRunOperator(trigger_dag_id="this-dag-id")

【讨论】：

是的，我也是这么做的。这对我有用。谢谢

【解决方案2】：

是的，您可以安排 DAG 每 2 分钟运行一次。

设置schedule_interval='*/2 * * * *'

Schedule_interval 接受 CRON 表达式：

https://en.wikipedia.org/wiki/Cron#CRON_expression

Structuring dag

如果您想在 2 分钟后连续重新运行，请尝试配置 TriggerDagRunOperator

【讨论】：

不，让我把问题说得更清楚些。我的 DAG 应该以这样的方式运行，即每次完成运行时，它都必须等待 2 分钟并重新开始运行。我不想安排我的 DAG 每 2 分钟运行一次，而是应该在同一 DAG 完成 2 分钟后立即连续运行
好的。然后尝试使用 TriggerDagRunOperator 重新运行 dag。 LINK
我想知道这种情况下的调度间隔是怎样的

【解决方案3】：

您需要在schedule_interval 中使用timedelta 作为：

from datetime import datetime, timedelta

default_args = {
    'owner': 'airflow',
    'start_date': datetime(2021, 5, 9),
}


with DAG(
    dag_id="my_dag",
    default_args=default_args,
    schedule_interval=timedelta(minutes=2),
    max_active_runs=1,
    catchup=False,
) as dag:
    your code...

在schedule_interval 中使用timedelta 意味着它将在上次运行完成后等待2 分钟，然后再触发下一次。

【讨论】：