airflow - 将 ds_add 和宏与 Airflow 一起使用

Question

我想得到 execution date - 2 months格式为 YYYY-MM-DD HH:MM

这是我写的:

EXEC_TIMESTAMP_2_MONTHS_AGO = "{{ macros.ds_add(execution_date.strftime('%Y-%m-%d/%H:%M'), -60) }}"

但是这不起作用。它不呈现。

问题是什么？

编辑:这是运算符(operator):

import_orders_op = MySqlToGoogleCloudStorageOperator(
    task_id='import_orders_and_upload_to_storage',
    mysql_conn_id='mysqlcon',
    google_cloud_storage_conn_id='gcpcon',
    provide_context=True,
    sql = 'import_orders.sql',
    params={'max_Orders_ID_2_months_ago': EXEC_TIMESTAMP_2_MONTHS_AGO, 'table_name' :  TABLE_NAME},
    bucket=GCS_BUCKET_ID,
    filename=file_name,
    dag=dag) 

错误是:

mysql_exceptions.ProgrammingError: (1064, "You have an error in your SQL syntax; check the manual that corresponds to your MySQL server version for the right syntax to use near '{ (execution_date - macros.timedelta(days=60)).strftime('%Y-%m-%d %H:%M') }}\n   ' at line 51")

Answer 1

理想情况下，您应该在帖子中包含日志中的任何错误消息，这将有助于 future 的搜索者，也有助于缩小问题范围。

我认为现在正在发生的事情是您正在使用 ds_add 但在 execution_date 上，这是一个合适的日期时间。通常它是在 ds 上完成的，它是一个字符串表示。

相反，您需要直接从 execution_date 中减去，然后然后按照您想要的方式对其进行格式化。

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

为了完整性 - 该计算只能在您的模板化 sql 或模板化 sql 文件中进行。