时间戳

首页 > 解决方案 > 如何将 BigQueryOperator 与 execution_date 一起使用？

airflow - 如何将 BigQueryOperator 与 execution_date 一起使用？

问题描述

这是我的代码：

EXEC_TIMESTAMP  = "{{  execution_date.strftime('%Y-%m-%d %H:%M')  }}"
query = """
        select ... where date_purchased between TIMESTAMP_TRUNC(cast ( {{ params.run_timestamp }} as TIMESTAMP), HOUR, 'UTC') ...
        """
generate_op = BigQueryOperator(
                    bql=query,
                    destination_dataset_table=table_name,
                    task_id='generate',
                    bigquery_conn_id=CONNECTION_ID,
                    use_legacy_sql=False,
                    write_disposition='WRITE_TRUNCATE',
                    create_disposition='CREATE_IF_NEEDED',
                    query_params={'run_timestamp': EXEC_TIMESTAMP},
                    dag=dag)

这应该工作，但它没有。渲染选项卡向我显示：

between TIMESTAMP_TRUNC(cast (  as TIMESTAMP), HOUR, 'UTC')

日期不见了。它被渲染成虚无。

我怎样才能解决这个问题？该运算符没有provide_context=True。我不知道该怎么办。

标签： airflow

解决方案

Luis，这query_params不是params您可以在模板上下文中引用的。它们没有添加到其中。由于params是空的，你{{ params.run_timestamp }}的要么是""要么None。如果您将其更改为params={'run_timestamp':…}它仍然会有问题，因为params值没有模板化。因此，当您使用模板化字段bql来包含时{{ params.run_timestamp }}，您将获得准确params: {'run_timestamp': …str… }填写的内容，而无需对该值进行任何递归扩展。你应该得到{{ execution_date.strftime('%Y-%m-%d %H:%M') }}.

让我尝试为您重新编写此代码（但我可能错误地将括号转换为错误，不确定）：

generate_op = BigQueryOperator(
                    sql="""
select ...
where date_purchased between
  TIMESTAMP_TRUNC(cast('{{execution_date.strftime('%Y-%m-%d %H:%M')}}') as TIMESTAMP), HOUR, 'UTC')
...
                    """,
                    destination_dataset_table=table_name,
                    task_id='generate',
                    bigquery_conn_id=CONNECTION_ID,
                    use_legacy_sql=False,
                    write_disposition='WRITE_TRUNCATE',
                    create_disposition='CREATE_IF_NEEDED',
                    dag=dag,
)

您可以看到bql和sql字段是模板化的。但是，该bql字段已被弃用并在以后的代码中被删除。

推荐阅读