Airflow BigQueryOperator: How to save output data to a specified partition column instead of ingestion time

Question

I like to pass partition col name to BigQueryOperator istead of using ingestion time partitioned tables.

        bq_cmd = BigQueryOperator (
            task_id=                    "task_id",
            sql=                        [query],
            destination_dataset_table=  destination_tbl,
            use_legacy_sql=             False,
            write_disposition=          'WRITE_TRUNCATE',
            time_partitioning=          {'time_partitioning_type':'DAY','time_partitioning_field': 'batch_date'},
            allow_large_results=        True,
            trigger_rule=               'all_success',
            query_params=               query_params,
            dag=                        dag
        )

I tried the above way but it did not work and the final table got created with partition col as _PARTITIONTIME instead of batch_date

Is destination_tbl already created (in other words the schema is set and when the big query operator runs it drops everything in the table but it is not creating a new table) or is it a new table each time this runs? — WIT
@WIT For first run , table gets created with ingestion time partition ,and with second run new partition gets created and so forth — Neha0908

Neha0908 Neha0908 · Accepted Answer · 2020-09-21T16:31:43

Pasing the params in below format , helps in creating partition table based on custom field

bq_cmd = BigQueryOperator (
    task_id=                    'task_name',
    sql=                        [query],
    destination_dataset_table=  destination_tbl,
    use_legacy_sql=             False,
    write_disposition=          'WRITE_TRUNCATE',
    time_partitioning=          {'field': 'execution_date',
                                 'type' : 'DAY'},
    allow_large_results=        True,
    trigger_rule=               'all_success',
    query_params=               query_params,
    dag=                        dag
)

In my case it was execution_date which was present in input table.

Airflow BigQueryOperator: How to save output data to a specified partition column instead of ingestion time

1 Answers