每天在特定时间运行 DAG

Question

我也多次阅读了有关schedule_interval、start_date 和 Airflow 文档的多个示例，但我仍然无法理解：

如何在每天的特定时间执行我的 DAG？例如，假设现在是上午 9:30（上午），我部署了我的 DAG，我希望它在 10:30 执行

我试过了

with DAG(
    "test",
    default_args=default_args,
    description= "test",
    schedule_interval = "0 10 * * *",
    start_date = days_ago(0),
    tags = ["goodie"]) as dag:

但由于某种原因，今天没有运行。我尝试了不同的start_dates altso start_date = datetime.datetime(2021,6,23)，但它没有被执行。

如果我将days_ago(0) 替换为days_ago(1)，它会一直落后1 天，即它今天没有运行，但昨天运行了

难道没有一种简单的方法可以说“我现在部署我的 DAG，我想用这个 cron 语法执行它”（我认为这是大多数人想要的）而不是计算执行时间，基于在start_date、schedule_interval 上并弄清楚，如何解释它？

Answer 1

如果我将 days_ago(0) 替换为 days_ago(1)，它总是落后 1 天

它不落后。您只是将 Airflow 调度机制与 cron 作业混淆了。在 cron 作业中，您只需提供一个 cron 表达式并相应地安排 - 这不是它在 Airflow 中的工作方式。

在 Airflow 中，调度由 start_date + schedule interval 计算。 Airflow 在间隔结束时执行作业。这与数据管道通常的工作方式是一致的。今天您正在处理昨天的数据，所以在这一天结束时您想要启动一个处理昨天记录的流程。

作为一项规则 - 永远不要使用动态开始日期。

设置：

with DAG(
    "test",
    default_args=default_args,
    description= "test",
    schedule_interval = "0 10 * * *",
    start_date = datetime(2021,06,23, 10 ,0), # 2021-06-23 10:00
    tags = ["goodie"]) as dag:

意味着第一个将在2021-06-24 10:00 开始，这次运行execution_date 将是2021-06-23 10:00。第二次运行将在2021-06-25 10:00 开始，这次运行execution_date 将是2021-06-24 10:00

由于这对许多新用户来说是一个困惑的根源，因此 AIP-39 Richer scheduler_interval 正在进行架构更改，这将在运行时间和运行时要考虑的间隔之间解耦 - 但正如提到的那样，这尚未最终确定。