google-bigquery - Airflow BigQueryOperator : how to save query result in a partitioned Table?

Question

我有一个简单的 DAG

from airflow import DAG
from airflow.contrib.operators.bigquery_operator import BigQueryOperator

with DAG(dag_id='my_dags.my_dag') as dag:

    start = DummyOperator(task_id='start')

    end = DummyOperator(task_id='end')
    sql = """
             SELECT *
             FROM 'another_dataset.another_table'
          """
    bq_query = BigQueryOperator(bql=sql,
                            destination_dataset_table='my_dataset.my_table20180524'),
                            task_id='bq_query',
                            bigquery_conn_id='my_bq_connection',
                            use_legacy_sql=False,
                            write_disposition='WRITE_TRUNCATE',
                            create_disposition='CREATE_IF_NEEDED',
                            query_params={})
    start >> bq_query >> end

执行 bq_query 时SQL 查询保存在分片表中的任务。我希望它保存在每日分区表中。为了做到这一点，我只改了 destination_dataset_table至 my_dataset.my_table$20180524 .执行 bq_task 时出现以下错误:

Partitioning specification must be provided in order to create partitioned table

如何指定 BigQuery 将查询结果保存到每日分区表？我的第一个猜测是使用 query_params在 BigQueryOperator但我没有找到有关如何使用该参数的任何示例。

编辑:

我正在使用 google-cloud==0.27.0 python 客户端……它是 Prod 中使用的那个 :(

Answer 1

您首先需要创建一个空分区目标表。按照此处的说明操作:link创建一个空的分区表

然后再次在 Airflow 管道下方运行。
你可以试试代码:

import datetime
from airflow import DAG
from airflow.contrib.operators.bigquery_operator import BigQueryOperator
today_date = datetime.datetime.now().strftime("%Y%m%d")
table_name = 'my_dataset.my_table' + '$' + today_date
with DAG(dag_id='my_dags.my_dag') as dag:
    start = DummyOperator(task_id='start')
    end = DummyOperator(task_id='end')
    sql = """
         SELECT *
         FROM 'another_dataset.another_table'
          """
    bq_query = BigQueryOperator(bql=sql,
                        destination_dataset_table={{ params.t_name }}),
                        task_id='bq_query',
                        bigquery_conn_id='my_bq_connection',
                        use_legacy_sql=False,
                        write_disposition='WRITE_TRUNCATE',
                        create_disposition='CREATE_IF_NEEDED',
                        query_params={'t_name': table_name},
                        dag=dag
                        )
start >> bq_query >> end

所以我所做的是我创建了一个动态表名变量并传递给了 BQ 操作符。