如何获取 execution_date 作为纪元时间？

Question

在常规的 python 代码中我可以做到：

import time
int(time.time())

这给了我作为时代的时间。 我希望能够使用气流宏来做到这一点：execution_date

这是我尝试过的：

"{{  strptime(execution_date.strftime('%Y-%m-%d %H:%M:%S'), '%d.%m.%Y %H:%M:%S') }}"

但这给出了：

jinja2.exceptions.UndefinedError: 'strptime' 未定义

我正在运行 Airflow 1.9 和 Python 2.7

Answer 1

从 Airflow 1.10 开始，Airflow 使用 Pendulum 作为日期时间，它的属性 timestamp()、int_timestamp 和 float_timestamp 返回纪元。

所以，你可以这样做：

{{ execution_date.int_timestamp }}

文档：https://pendulum.eustace.io/docs/#attributes-and-properties

其他选项有：

{{ execution_date.strftime('%s') }} # Also Python 2
{{ execution_date.timestamp() }}    # Python >=3.3

Answer 2

有几种方法可以解决这个问题，但首先，您应该简化访问时间戳的方式。

当您使用 Airflow 时，其中 execution_date 是一个 pendulum.Pendulum 对象，您可以访问 execution_date.int_timestamp 属性。

在更一般的情况下，假设您使用的是 Python 3.3+，有一种更简单的方法可以使用 datetime 对象上的 timestamp() 方法（也可以在 Pendulum 对象上使用）来获取纪元时间戳。

execution_date.int_timestamp
Output => 1555477775

Python 3.3+:
int(execution_date.timestamp())
Output => 1555477775

Python 2.7+:
int(v.strftime('%s'))
Output => 1555477775

1。在模板内直接格式化

此方法使用 datetime 对象的 timestamp() 方法生成一个浮点数，然后通过 Jinja2 模板的内置 int filter 运行。

For Pendulum:
{{ execution_date.int_timestamp }}

For general datetime objects:
{{ execution_date.timestamp()|int }}

2。创建自定义 Jinja2 过滤器

如果您在整个模板中的多个位置使用此转换，则可能值得创建一个过滤器来集中它。使用 Airflow，您可以在 DAG 对象上register a custom filter。

import time

def epoch_int(v):
    return int(v.strftime('%s'))

dag = DAG(
  ...
  user_defined_filters={
    "epoch_int": epoch_int,
  }
  ...
)

然后在你的模板中使用它。

{{ execution_date|epoch_int }}