Airflow : dag run with execution_date = trigger_date = fixed_schedule

Question

在 Airflow 中，我想每周一上午 8 点运行一次 dag(execution_date 当然应该是“当天星期一上午 8 点”)。为此工作流程设置的相关参数是:

• 开始日期:“2018-03-19”
• schedule_interval:“0 8 * * 周一”

我希望每周一早上 8 点都能看到一场 dag 运行事件。第一个运行时间为 2018 年 3 月 19 日上午 8 点，execution_date = 2018-03-19-08-00-00，每周一依此类推。

然而事实并非如此:dag 并未于 2018 年 3 月 19 日上午 8 点开始。此处解释了真实的行为，例如:https://stackoverflow.com/a/39620901/1510109或https://stackoverflow.com/a/48213964/1510109行为是:在间隔的每个结束处(在我的例子中是每周)，dag 以execution_date = 间隔的开始(即前一周)运行。这种行为显然是由“ETL 思维方式”驱动的(参见上面的链接)。但这绝对不是我想要的。

如何才能在每周一上午 08:00 使用 execution_date = trigger_date = now(= 当前周一上午 8 点)运行我的 dag？

谢谢

Answer 1

拿一个quick look at my answer包含开始时间和执行日期示例。

您想每周一早上 8 点运行。

所以这部分将保持不变:

schedule_interval: '0 8 * * MON',

您希望它在 2018 年 3 月 19 日首次运行，因为首次运行发生在开始日期后的第一个完整计划周期结束时，您应该将开始日期更改为:

start_date: datetime(2018,03,12),

您必须接受这样一个事实:Airflow 会以每个周期的开始时间来命名您的 DagRun，并根据设置为间隔周期开始时间的 execution_date 传入宏。相应地调整你的逻辑。

您的第一次运行将在 2018-03-19T08:00:00.0Z 和 execution_date、依赖于它的所有其他宏以及 DagRun 的名称之后开始将是 2018-03-12T08:00:00.0Z

只要您了解 execution_date 的预期结果，并且您不尝试根据 datetime.now() 确定时间，您的 DAG 就能够在操作上是幂等的。您可以在任何 PythonOperator 或自定义运算符中随意创建一个新变量，例如 my_execution_date =execution_date + datetime.timedelta(7)(您可以从任务上下文中获取execution_date)，使用模板语句，例如 {{ (execution_date + Macros.timedelta(7)).strftime('%Y%m%d') }} 或 {{ macros.ds_add(ds, 7) ) }}，或使用next_execution_date。

您甚至可以添加 dag 级别 user_define_macros，例如 {'dt':lambda d: d+datetime.timedelta(days=7)} 来启用 {{ dt(execution_date) }}。最近添加了 user_define_filters，例如 {'dt':lambda d: d+datetime.timedelta(days=7)} 启用 {{execution_date | dt}}。 next_ds 和 next_execution_date 会更容易满足您的目的。

在考虑模板时，您不妨阅读一下内置的内容:http://jinja.pocoo.org/docs/2.10/templates/#builtin-filters