22

最近我测试了很多气流，execution_date运行时有一个问题airflow trigger_dag <my-dag>。

我了解到这execution_date不是我们第一次从这里想到的：

Airflow 是作为 ETL 需求的解决方案而开发的。在 ETL 世界中，您通常会汇总数据。因此，如果我想汇总 2016 年 2 月 19 日的数据，我会在格林威治标准时间 2016 年 2 月 20 日午夜进行，这将是在 2016 年 2 月 19 日的所有数据可用之后。

start_date = datetime.combine(datetime.today(),
                              datetime.min.time())

args = {
    "owner": "xigua",
    "start_date": start_date
}
dag = DAG(dag_id="hadoopprojects", default_args=args,
          schedule_interval=timedelta(days=1))


wait_5m = ops.TimeDeltaSensor(task_id="wait_5m",
                              dag=dag,
                              delta=timedelta(minutes=5))

上面的代码是我日常工作流程的开始部分，第一个任务是 TimeDeltaSensor，它在实际工作前再等 5 分钟，所以这意味着我的 dag 将在2016-09-09T00:05:00, 2016-09-10T00:05:00... 等处触发。

在 Web UI 中，我可以看到类似的scheduled__2016-09-20T00:00:00内容，并且任务运行在2016-09-21T00:00:00，根据ETL模型，这似乎是合理的。

但是有一天我的 dag 没有因未知原因被触发，所以我手动触发它，如果我在 触发它2016-09-20T00:10:00，那么 TimeDeltaSensor 将等到2016-09-21T00:15:00运行之前。

这不是我想要的，我希望它2016-09-20T00:15:00不是在第二天运行，我尝试过execution_date通过--conf '{"execution_date": "2016-09-20"}'，但它不起作用。

我应该如何处理这个问题？

$ airflow version
[2016-09-21 17:26:33,654] {__init__.py:36} INFO - Using executor LocalExecutor
  ____________       _____________
 ____    |__( )_________  __/__  /________      __
____  /| |_  /__  ___/_  /_ __  /_  __ \_ | /| / /
___  ___ |  / _  /   _  __/ _  / / /_/ /_ |/ |/ /
 _/_/  |_/_/  /_/    /_/    /_/  \____/____/|__/
   v1.7.1.3
4

2 回答 2

54

首先，我建议您使用常量 for start_date，因为动态常量会根据调度程序评估的气流管道做出不可预测的行为。

有关start_date此处的更多信息，请参阅我编写的常见问题解答条目并将所有内容整理出来： https ://airflow.apache.org/faq.html#what-s-the-deal-with-start-date

现在，execution_date当它被触发时，这是人们在 Airflow 上的常见问题。气流设置execution_date基于它所覆盖的计划周期的左边界，而不是基于它何时触发（这将是周期的右边界）。例如，在运行schedule='@hourly'任务时，任务将每小时触发一次。在下午 2 点触发的任务将有一个execution_date下午 1 点，因为它假设您在下午 2 点处理下午 1 点到下午 2 点的时间窗口。同样，如果您运行日常作业，运行 with execution_dateof2016-01-01将在午夜后不久触发2016-01-02。

在考虑 ETL 和差分负载时，这种左边界标记很有意义，但在考虑简单的、类似 cron 的调度程序时会变得混乱。

于 2016-09-21T15:34:49.627 回答
2

Airflow 将提供 UTC 时间。我不确定您在哪个时区运行任务。因此，请确保您考虑 UTC 时区并相应地安排或触发作业。

尝试将您要触发的时间转换为 UTC 时间并触发 DAG。有用。有关更多信息，您可以阅读以下链接

https://cwiki.apache.org/confluence/display/AIRFLOW/Common+Pitfalls

于 2017-11-30T14:56:11.157 回答