从今天即 2019-12-18 开始，在气流中安排一个 DAG 每 5 分钟运行一次答案

【问题标题】：Schedule a DAG in airflow to run for every 5 minutes , starting from today i.e., 2019-12-18从今天即 2019-12-18 开始，在气流中安排一个 DAG 每 5 分钟运行一次
【发布时间】：2019-12-18 11:30:41
【问题描述】：

我正在尝试从今天（2019-12-18）开始每 5 分钟运行一次 DAG。我将开始日期定义为 start_date:dt.datetime(2019, 12, 18, 10, 00, 00) 并将计划间隔定义为 schedule_interval= '*/5 * * * *' 。当我启动airflow scheduler 时，我没有看到我的任何任务正在运行。

但是当我将start_date 修改为start_date:dt.datetime(2019, 12, 17, 10, 00, 00) 即昨天的日期时，DAG 会连续运行 10 秒而不是 5 分钟。

我认为解决这个问题的方法是正确设置start_date，但我找不到完美的解决方案。请帮帮我！

这是我的代码。

from airflow import DAG
from airflow.operators.bash_operator import BashOperator
import datetime as dt
from airflow.operators.python_operator import PythonOperator

def print_world():
   print('world')


default_args = {
    'owner': 'bhanuprakash',
    'depends_on_past': False,
    'start_date': dt.datetime(2019, 12, 18, 10, 00, 00),
    'email': ['bhanuprakash.uchula@techwave.net'],
    'email_on_failure': False,
    'email_on_retry': False,
    'retries': 1,
    'retry_delay': dt.timedelta(minutes=5)
}

with DAG('dag_today',
    default_args=default_args,
    schedule_interval= '*/5 * * * *'
    ) as dag:


    print_hello = BashOperator(task_id='print_hello',
        bash_command='gnome-terminal')


    sleep = BashOperator(task_id='sleep',
        bash_command='sleep 5')


    print_world = PythonOperator(task_id='print_world',
        python_callable=print_world)

print_hello >> sleep >> print_world

【问题讨论】：

标签： python airflow directed-acyclic-graphs airflow-scheduler

【解决方案1】：

您传递给 Airflow 的日期时间对象不支持时区。 Airflow 在内部使用 UTC。您传递给 Airflow 的天真日期时间对象可能与调度程序的时间概念不一致，这可能是 DAG 没有被安排在“今天”午夜运行的原因 (2019-12-18)。

而不是像这样传递一个简单的日期时间对象：

'start_date': dt.datetime(2019, 12, 18, 10, 00, 00)

尝试使用 pendulum 让您的 DAG 时区感知：

import pendulum

...
'start_date': pendulum.datetime(year=2019, month=12, day=10).astimezone('YOUR TIMEZONE'), # See list of tz database time zones here -> https://en.wikipedia.org/wiki/List_of_tz_database_time_zones

文档 (https://airflow.apache.org/docs/stable/timezone.html) 非常有用，可以获得有关如何在 Airflow 中处理日期时间的提示。

至于您关于运行频率的其他问题......默认情况下，DAG 运行旨在对您的开始日期和结束日期之间的所有时间间隔进行“追赶”。要禁用此行为，您需要在实例化 DAG 时添加 catchup=False。

来自Airflow docs

回填和追赶

一个带有 start_date、可能是 end_date 和一个 schedule_interval 定义了调度程序的一系列间隔变成单独的 Dag Runs 并执行。 Airflow 的一项关键能力是这些 DAG 运行是原子的、幂等的项目，并且默认情况下，调度程序将检查 DAG 的生命周期（从开始到结束/现在，一次一个间隔）并开始 DAG 运行任何尚未运行（或已清除）的间隔。这个概念称为追赶。

如果您的 DAG 是为处理自己的追赶而编写的（IE 不限于间隔，而是改为“现在”。），那么你会想要关闭追赶（在 DAG 本身上使用 dag.catchup = False) 或默认在配置文件级别使用 catchup_by_default = 假。这样做的目的是指示调度程序只为最新的实例创建 DAG 运行 DAG 区间序列。

我建议浏览我链接的两个页面，以更好地了解基本的 Airflow 概念。

【讨论】：