How to format macros in Airflow?

user_10163853 0
Viewed 12304

I have the following:

EXEC_DATE1 = '{{ macros.ds_add(ds, 1) }}'


EXEC_DATE2 = '{{ execution_date }}'

I want to create path variable that looks like:

path1 = EXEC_DATE1 + '/' + HH:MM (of EXEC_DATE1)
path2 = EXEC_DATE2 + '/' + HH:MM (of EXEC_DATE2)

Eventually it should be like:

2018-09-16/10:41

How can I do that?

I tried:

EXEC_DATE = '{{ execution_date }}'
EXEC_DATE = EXEC_DATE.strftime('%Y-%m-%d/%H:%M')

but it gives:

'str' object has no attribute 'strftime'

EDIT: My code:

EXEC_TIMESTAMP_PATH = "{{  execution_date.strftime('%Y-%m-%d/%H:%M') }}"
EXEC_DATE = "{{  execution_date.strftime('%H:%M') }}"
EXEC_TIME = "{{  mexecution_date.strftime('%Y-%m-%d') }}"

task3_op= BashOperator(
    task_id='task3',
    params={'EXEC_DATE':EXEC_DATE, 'EXEC_TIME':EXEC_TIME},
    bash_command="""python3 script.py '{{ var.value.task3_variable }}' '{{ params.EXEC_DATE }}' '{{ params.EXEC_TIME }}' 'file.json'""",
    dag=dag)

This doesn't work. The parameters are not rendered.

airflow
3 Answers

You need to do it as follows:

EXEC_DATE = "{{ execution_date.strftime('%Y-%m-%d/%H:%M') }}"

The strftime should be used inside the curly braces.

If you need next_execution date, use the following:

EXEC_DATE = "{{ next_execution_date.strftime('%Y-%m-%d/%H:%M') }}"

And if you just want to add a timedelta:

EXEC_DATE = "{{ (execution_date + macros.timedelta(days=1)).strftime('%Y-%m-%d/%H:%M') }}"

You code can be as follows:

BASH_COMMAND="""
python3 script.py {{ var.value.task3_variable }} {{  execution_date.strftime('%H:%M') }} {{  execution_date.strftime('%Y-%m-%d') }} file.json
"""

task3_op= BashOperator(
    task_id='task3',
    bash_command=BASH_COMMAND,
    dag=dag)
user_56915250

Specifically what's not working in your question's code is that your params don't have templating (macro expansion) run on them, but your bash_command does. So I'm betting the rendered template view of the bash_command is like E.G.:

python3 script.py 'task3_variable's value' '{{  execution_date.strftime('%H:%M') }}' '{{  execution_date.strftime('%Y-%m-%d') }}' 'file.json'

That is not getting expanded any further, and the ' are probably throwing the meaning off for Bash too, because effectively you have:

params = { 'EXEC_DATE': "{{  execution_date.strftime('%H:%M') }}",
           'EXEC_TIME': "{{  execution_date.strftime('%Y-%m-%d') }}" }

So removing one level of macro expansion by directly using the templates you had set in the params in the bash_command instead, is working better for you.

task3_op = BashOperator(
    task_id='task3',
    bash_command="""
python3 script.py '{{ var.value.task3_variable }}' \
'{{execution_date.strftime('%H:%M')}}' '{{execution_date.strftime('%Y-%m-%d')}}' file.json
    """,
    dag=dag)

You should use the airflow render command or check the rendered view in the task instance details of the UI when debugging issues like this.

It would likely show you what's going wrong with your assumed macro expansions.
This example should render to something like:

python3 script.py 'task3_variable's value' \
'12:21' '2018-09-16' file.json

You were hoping to make two paths based on dates; try:

task3_op= BashOperator(
    task_id='task3',
    bash_command="""
python3 script.py \
  '{{ var.value.task3_variable }}' \
  '{{(execution_date + macros.timedelta(days=1).strftime('%Y-%m-%d/%H:%M')}}' \
  '{{execution_date.strftime('%Y-%m-%d/%H:%M')}}' \
  file.json
    """,
    dag=dag)

Airflow 1.10 also added next_execution_date which might work better for you than adding a day to execution_date if your interval is daily.

dlamblin_4590

Incase in your code macros aren't rendering in your custom Airflow Operator, do add this to your custom class function before init

template_fields = ('name_of_function_you_passed_macro_in')

user_73914720
Related