如何在 Airflow 中格式化宏？答案

【问题标题】：How to format macros in Airflow?如何在 Airflow 中格式化宏？
【发布时间】：2018-09-16 09:47:53
【问题描述】：

我有以下几点：

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


EXEC_DATE2 = '{{ execution_date }}'

我想创建如下所示的路径变量：

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

最终应该是这样的：

2018-09-16/10:41

我该怎么做？

我试过了：

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

但它给出了：

'str' object has no attribute 'strftime'

编辑： 我的代码：

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)

这不起作用。参数不渲染。

【问题讨论】：

标签： airflow

【解决方案1】：

你需要这样做：

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

strftime 应该用在花括号内。

如果您需要下一个执行日期，请使用以下内容：

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

如果你只想添加一个timedelta:

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

你的代码可以如下：

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)

【讨论】：

它不是为我的 bashoperator 渲染的

【解决方案2】：

具体而言，在您的问题代码中不起作用的是您的 params 没有在其上运行模板（宏扩展），但您的 bash_command 可以。所以我打赌bash_command 的渲染模板视图就像E.G.：

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

这并没有进一步扩展，' 可能也对 Bash 失去了意义，因为实际上你有：

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

因此，通过直接使用您在params 中设置的模板在bash_command 中删除一级宏扩展对您来说效果更好。

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)

在调试此类问题时，您应该使用airflow render command 或检查 UI 的任务实例详细信息中的渲染视图。

它可能会告诉您假设的宏扩展出了什么问题。
此示例应呈现为：

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

您希望根据日期创建两条路径；试试：

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 还添加了next_execution_date，如果您的间隔是每天，这可能比在execution_date 中添加一天更适合您。

【讨论】：

【解决方案3】：

如果您的代码宏未在您的自定义 Airflow Operator 中呈现，请务必在 init

之前将此添加到您的自定义类函数中

template_fields = ('name_of_function_you_passed_macro_in')

【讨论】：