Is there an operator in Airflow to create a table from a query in BigQuery?

user_8248194 0
Viewed 912

I'm looking for something like

CreateBQTableOperator(
    query='select * from my_table',
    output_table='my_other_table'
)

I'm looking for either an already existing operator or code for such an operator. The operator should take another argument to decide if deleting the table (if the table exists) before recreating it or appending the query to the current table.

airflow google-bigquery python
1 Answers

For Airflow >= 1.10 with providers you can use BigQueryInsertJobOperator This operator is using JobConfigurationQuery you can configure any option supported by the API using configuration parameter:

from airflow.providers.google.cloud.operators.bigquery import BigQueryInsertJobOperator

execute_query_save = BigQueryInsertJobOperator(
    task_id="execute_query_save",
    configuration={
        "query": {
            "query": "select * from my_table",
            "useLegacySql": False,
            "writeDisposition": "WRITE_EMPTY",
            'destinationTable': {
                'projectId': "my-project",
                'datasetId': "my_data_set",
                'tableId': "table2"
            },
        }
    },
)

For older Airflow versions you can use BigQueryExecuteQueryOperator:

The operator has destination_dataset_table:

from airflow.providers.google.cloud.operators.bigquery import BigQueryExecuteQueryOperator

execute_query_save = BigQueryExecuteQueryOperator(
    task_id="execute_query_save",
    sql="SELECT * FROM my_data_set.table1",
    use_legacy_sql=False,
    destination_dataset_table="my_data_set.table2",
    location="southamerica-east1",
    write_disposition="WRITE_EMPTY",
    create_disposition="CREATE_IF_NEEDED",
)

You can control over the requested behavior with setting the values for the parameters (reference for values from Google docs).

write_disposition options are:

WRITE_TRUNCATE: If the table already exists, BigQuery overwrites the table data and uses the schema from the query result.

WRITE_APPEND: If the table already exists, BigQuery appends the data to the table.

WRITE_EMPTY: If the table already exists and contains data, a 'duplicate' error is returned in the job result.

create_disposition options are:

CREATE_IF_NEEDED: If the table does not exist, BigQuery creates the table.

CREATE_NEVER: The table must already exist. If it does not, a 'notFound' error is returned in the job result.

user_146244090
Related