4

In regular python code I can do:

import time
int(time.time())

This gives me the time as epoch. I want to be able to do this with airflow macro: execution_date

This is what I tried:

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

But this gives:

jinja2.exceptions.UndefinedError: 'strptime' is undefined

I'm running Airflow 1.9 & Python 2.7

KoKu
  • 43
  • 1
  • 4
  • You should do the datetime conversion in the Python code first and store it's value in a variable and then render the variable in the Jinja template – amanb Apr 17 '19 at 05:55
  • Have a look at the [documentation](http://jinja.pocoo.org/docs/2.10/api/#custom-filters) on custom filters – cullzie Apr 17 '19 at 06:01

2 Answers2

5

Since Airflow 1.10, Airflow uses Pendulum for datetimes, which has attributes timestamp(), int_timestamp and float_timestamp which return the epoch.

So, you could do:

{{ execution_date.int_timestamp }}

Docs: https://pendulum.eustace.io/docs/#attributes-and-properties

Other options are:

{{ execution_date.strftime('%s') }} # Also Python 2
{{ execution_date.timestamp() }}    # Python >=3.3
Bas Harenslak
  • 2,591
  • 14
  • 14
2

There are several approaches to solving this, but first, you should simplify how you are accessing the timestamp.

As you are using Airflow, where execution_date is a pendulum.Pendulum object, you have access to the execution_date.int_timestamp property.

In a more general situation assuming you are using Python 3.3+ there is a simpler way to get the epoch timestamp using the timestamp() method on a datetime object (which is also available on a Pendulum object).

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. Direct formatting within the template

This method uses the timestamp() method of the datetime object to generate a float which then run through the built-in int filter of the Jinja2 template.

For Pendulum:
{{ execution_date.int_timestamp }}

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

2. Create a custom Jinja2 Filter

If you are using this transformation in multiple places throughout your templates, it is probably worth creating a filter to centralise this. With Airflow, you can register a custom filter on the DAG object.

import time

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

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

and then use it in your template.

{{ execution_date|epoch_int }}
Steve McCartney
  • 191
  • 1
  • 3