airflow - DagRun 的 execution_date 是如何设置的？

Question

给定一个 DAG 有一个 start_date，它在特定日期运行，相应的 DAGRun 的 execution_date 是怎样的 定义了吗？

我已阅读 documentation但是一个例子让我感到困惑:

"""
Code that goes along with the Airflow tutorial located at:
https://github.com/airbnb/airflow/blob/master/airflow/example_dags/tutorial.py
"""
from airflow import DAG
from airflow.operators.bash_operator import BashOperator
from datetime import datetime, timedelta


default_args = {
    'owner': 'airflow',
    'depends_on_past': False,
    'start_date': datetime(2015, 12, 1),
    'email': ['airflow@example.com'],
    'email_on_failure': False,
    'email_on_retry': False,
    'retries': 1,
    'retry_delay': timedelta(minutes=5),
    'schedule_interval': '@hourly',
}

dag = DAG('tutorial', catchup=False, default_args=default_args)

假设 DAG 在 2016-01-02 早上 6 点运行，第一个 DAGRun 的 execution_date 为 2016- 01-01 并且如文档中所述

the next one will be created just after midnight on the morning of 2016-01-03 with an execution date of 2016-01-02

这是我将如何设置 execution_date:

DAG 将其 schedule_interval 设置为每小时并在 2016 年 1 月 2 日早上 6 点运行，execution_date第一个 DAGRun 将设置为 2016-01-02 早上 7 点，第二个设置为 2016-01-02 早上 8 点......等等。

Answer 1

这就是 Airflow 中调度的工作原理。当您考虑正常的 ETL 批处理过程如何运行以及如何使用 execution_date 来获取已更改的增量记录时，我认为按照 Airflow 的方式进行操作是有意义的。

假设我们想要安排一个批处理作业每晚运行以从某个源数据库中提取新记录。我们想要从 2018 年 1 月 1 日起更改的所有记录(我们也希望所有记录在 1 日更改)。为此，您可以将 DAG 的开始日期设置为 2018 年 1 月 1 日，调度程序将运行多次，但是当它到达 2018 年 2 月 1 日(或之后不久)时，它将运行我们的 DAG 执行日期，2018 年 1 月 1 日。

现在我们可以使用 JINJA 模板将 SQL 语句发送到源数据库，它使用 execution_date 作为 SQL 的一部分。 SQL 看起来像这样:

SELECT row1, row2, row3
FROM table_name
WHERE timestamp_col >= {{ execution_date }} and timestamp_col < {{ next_execution_date }}

我认为当你以这种方式看待它时它更有意义，尽管我承认我在一开始试图理解它时遇到了困难。

这里引用自文档 https://airflow.apache.org/scheduler.html :

The scheduler runs your job one schedule_interval AFTER the start date, at the END of the period.

另外值得注意的是，您从文档中查看的示例描述了禁用回填时计划的行为。如果启用回填，如果 DAG 之前从未运行过，那么从 2015 年 12 月到当前日期之间每隔 1 小时就会创建一次 DAG 运行。