Google Cloud BigQuery Operators¶

BigQuery is Google’s fully managed, petabyte scale, low cost analytics data warehouse. It is a serverless Software as a Service (SaaS) that doesn’t need a database administrator. It allows users to focus on analyzing data to find meaningful insights using familiar SQL.

Airflow provides operators to manage datasets and tables, run queries and validate data.

Note

GoogleSQL is the recommended dialect for BigQuery. BigQuery legacy SQL availability is restricted after June 1, 2026, based on legacy SQL usage during Google’s evaluation period. In Airflow, the implicit default for older BigQuery operators that expose use_legacy_sql is deprecated and will change from True to False in a future provider release. Set use_legacy_sql=True explicitly if you still need legacy SQL, or set use_legacy_sql=False to use GoogleSQL. For more information, see Legacy SQL feature availability.

Prerequisite Tasks¶

To use these operators, you must do a few things:

Select or create a Cloud Platform project using the Cloud Console.

Enable billing for your project, as described in the Google Cloud documentation.

Enable the API, as described in the Cloud Console documentation.
Install API libraries via pip.
pip install 'apache-airflow[google]'
Detailed information is available for Installation.
Setup a Google Cloud Connection.

Manage datasets¶

Create dataset¶

To create an empty dataset in a BigQuery database you can use BigQueryCreateEmptyDatasetOperator.

tests/system/google/cloud/bigquery/example_bigquery_dataset.py[source]

create_dataset = BigQueryCreateEmptyDatasetOperator(task_id="create_dataset", dataset_id=DATASET_NAME)

Get dataset details¶

To get the details of an existing dataset you can use BigQueryGetDatasetOperator.

This operator returns a Dataset Resource.

tests/system/google/cloud/bigquery/example_bigquery_dataset.py[source]

get_dataset = BigQueryGetDatasetOperator(task_id="get-dataset", dataset_id=DATASET_NAME)

List tables in dataset¶

To retrieve the list of tables in a given dataset use BigQueryGetDatasetTablesOperator.

tests/system/google/cloud/bigquery/example_bigquery_tables.py

get_dataset_tables = BigQueryGetDatasetTablesOperator(
    task_id="get_dataset_tables", dataset_id=DATASET_NAME
)

Update table¶

To update a table in BigQuery you can use BigQueryUpdateTableOperator.

The update method replaces the entire Table resource, whereas the patch method only replaces fields that are provided in the submitted Table resource.

tests/system/google/cloud/bigquery/example_bigquery_tables.py

update_table = BigQueryUpdateTableOperator(
    task_id="update_table",
    dataset_id=DATASET_NAME,
    table_id="test_table",
    fields=["friendlyName", "description"],
    table_resource={
        "friendlyName": "Updated Table",
        "description": "Updated Table",
    },
)

You can use this operator to update a view.

tests/system/google/cloud/bigquery/example_bigquery_tables.py

update_view = BigQueryUpdateTableOperator(
    task_id="update_view",
    dataset_id=DATASET_NAME,
    table_id="test_view",
    fields=["friendlyName", "description"],
    table_resource={
        "friendlyName": "Updated View friendlyName",
        "description": "Updated View description",
    },
)

And use the same operator to update a materialized view.

tests/system/google/cloud/bigquery/example_bigquery_tables.py

update_materialized_view = BigQueryUpdateTableOperator(
    task_id="update_materialized_view",
    dataset_id=DATASET_NAME,
    table_id="test_materialized_view",
    fields=["friendlyName", "description"],
    table_resource={
        "friendlyName": "Updated View friendlyName",
        "description": "Updated View description",
    },
)

Update dataset¶

To update a dataset in BigQuery you can use BigQueryUpdateDatasetOperator.

The update method replaces the entire dataset resource, whereas the patch method only replaces fields that are provided in the submitted dataset resource.

tests/system/google/cloud/bigquery/example_bigquery_dataset.py[source]

update_dataset = BigQueryUpdateDatasetOperator(
    task_id="update_dataset",
    dataset_id=DATASET_NAME,
    dataset_resource={"description": "Updated dataset"},
)

Delete dataset¶

To delete an existing dataset from a BigQuery database you can use BigQueryDeleteDatasetOperator.

tests/system/google/cloud/bigquery/example_bigquery_dataset.py[source]

delete_dataset = BigQueryDeleteDatasetOperator(
    task_id="delete_dataset", dataset_id=DATASET_NAME, delete_contents=True
)

Manage tables¶

Create table¶

To create a new table in a dataset with the data in Google Cloud Storage you can use BigQueryCreateTableOperator by providing table structure in table_resource field.

tests/system/google/cloud/bigquery/example_bigquery_tables.py

create_table = BigQueryCreateTableOperator(
    task_id="create_table",
    dataset_id=DATASET_NAME,
    table_id="test_table",
    table_resource={
        "schema": {
            "fields": [
                {"name": "emp_name", "type": "STRING", "mode": "REQUIRED"},
                {"name": "salary", "type": "INTEGER", "mode": "NULLABLE"},
            ],
        },
    },
)

You can also specify Google Cloud Storage object name as a way to specify schema. The object in Google Cloud Storage must be a JSON file with the schema fields in it.

tests/system/google/cloud/bigquery/example_bigquery_tables.py

create_table_schema_json = BigQueryCreateTableOperator(
    task_id="create_table_schema_json",
    dataset_id=DATASET_NAME,
    table_id="test_table",
    gcs_schema_object=GCS_PATH_TO_SCHEMA_JSON,
    table_resource={
        "tableReference": {
            "projectId": PROJECT_ID,
            "datasetId": DATASET_NAME,
            "tableId": "test_table",
        },
    },
)

You can use this operator to create a view on top of an existing table.

tests/system/google/cloud/bigquery/example_bigquery_tables.py

create_view = BigQueryCreateTableOperator(
    task_id="create_view",
    dataset_id=DATASET_NAME,
    table_id="test_view",
    table_resource={
        "view": {
            "query": f"SELECT * FROM `{PROJECT_ID}.{DATASET_NAME}.test_table`",
            "useLegacySql": False,
        },
    },
)

You can also use this operator to create a materialized view that periodically caches results of a query for increased performance and efficiency.

tests/system/google/cloud/bigquery/example_bigquery_tables.py

create_materialized_view = BigQueryCreateTableOperator(
    task_id="create_materialized_view",
    dataset_id=DATASET_NAME,
    table_id="test_materialized_view",
    table_resource={
        "materializedView": {
            "query": f"SELECT SUM(salary) AS sum_salary FROM `{PROJECT_ID}.{DATASET_NAME}.test_table`",
            "enableRefresh": True,
            "refreshIntervalMs": 600000,
        },
    },
)

Fetch data from table¶

To fetch data from a BigQuery table you can use BigQueryGetDataOperator . Alternatively you can fetch data for selected columns if you pass fields to selected_fields.

Note

The project_id parameter is deprecated and will be removed in a future release. Please use table_project_id instead.

The result of this operator can be retrieved in two different formats based on the value of the as_dict parameter: False (default) - A Python list of lists, where the number of elements in the nesting list will be equal to the number of rows fetched. Each element in the nesting will a nested list where elements would represent the column values for that row. True - A Python list of dictionaries, where each dictionary represents a row. In each dictionary, the keys are the column names and the values are the corresponding values for those columns.

tests/system/google/cloud/bigquery/example_bigquery_queries.py

get_data = BigQueryGetDataOperator(
    task_id="get_data",
    dataset_id=DATASET_NAME,
    table_id=TABLE_1,
    max_results=10,
    selected_fields="value,name",
    use_legacy_sql=False,
)

The below example shows how to use BigQueryGetDataOperator in async (deferrable) mode. Note that a deferrable task requires the Triggerer to be running on your Airflow deployment.

tests/system/google/cloud/bigquery/example_bigquery_queries_async.py

get_data = BigQueryGetDataOperator(
    task_id="get_data",
    dataset_id=DATASET_NAME,
    table_id=TABLE_NAME_1,
    use_legacy_sql=False,
    max_results=10,
    selected_fields="value",
    location=LOCATION,
    deferrable=True,
)

Upsert table¶

To upsert a table you can use BigQueryUpsertTableOperator.

This operator either updates the existing table or creates a new, empty table in the given dataset.

tests/system/google/cloud/bigquery/example_bigquery_tables.py

upsert_table = BigQueryUpsertTableOperator(
    task_id="upsert_table",
    dataset_id=DATASET_NAME,
    table_resource={
        "tableReference": {"tableId": "test_table_id"},
        "expirationTime": (int(time.time()) + 300) * 1000,
    },
)

Update table schema¶

To update the schema of a table you can use BigQueryUpdateTableSchemaOperator.

This operator updates the schema field values supplied, while leaving the rest unchanged. This is useful for instance to set new field descriptions on an existing table schema.

tests/system/google/cloud/bigquery/example_bigquery_tables.py

update_table_schema = BigQueryUpdateTableSchemaOperator(
    task_id="update_table_schema",
    dataset_id=DATASET_NAME,
    table_id="test_table",
    schema_fields_updates=[
        {"name": "emp_name", "description": "Name of employee"},
        {"name": "salary", "description": "Monthly salary in USD"},
    ],
)

Delete table¶

To delete an existing table you can use BigQueryDeleteTableOperator.

tests/system/google/cloud/bigquery/example_bigquery_tables.py

delete_table = BigQueryDeleteTableOperator(
    task_id="delete_table",
    deletion_dataset_table=f"{PROJECT_ID}.{DATASET_NAME}.test_table",
)

You can also use this operator to delete a view.

tests/system/google/cloud/bigquery/example_bigquery_tables.py

delete_view = BigQueryDeleteTableOperator(
    task_id="delete_view",
    deletion_dataset_table=f"{PROJECT_ID}.{DATASET_NAME}.test_view",
)

You can also use this operator to delete a materialized view.

tests/system/google/cloud/bigquery/example_bigquery_tables.py

delete_materialized_view = BigQueryDeleteTableOperator(
    task_id="delete_materialized_view",
    deletion_dataset_table=f"{PROJECT_ID}.{DATASET_NAME}.test_materialized_view",
)

Manage routines¶

Airflow exposes the BigQuery routines API (user-defined functions, stored procedures, and table-valued functions) through a small set of dedicated operators and a sensor. See Google Cloud BigQuery Routines Operators for the full guide with examples for each routine type.

Execute BigQuery jobs¶

Let’s say you would like to execute the following query.

tests/system/google/cloud/bigquery/example_bigquery_queries.py

INSERT_ROWS_QUERY = (
    f"INSERT {DATASET_NAME}.{TABLE_1} VALUES "
    f"(42, 'monty python', '{INSERT_DATE}'), "
    f"(42, 'fishy fish', '{INSERT_DATE}');"
)

To execute the SQL query in a specific BigQuery database you can use BigQueryInsertJobOperator with proper query job configuration that can be Jinja templated.

tests/system/google/cloud/bigquery/example_bigquery_queries.py

insert_query_job = BigQueryInsertJobOperator(
    task_id="insert_query_job",
    configuration={
        "query": {
            "query": INSERT_ROWS_QUERY,
            "useLegacySql": False,
            "priority": "BATCH",
        }
    },
)

The below example shows how to use BigQueryInsertJobOperator in async (deferrable) mode. Note that a deferrable task requires the Triggerer to be running on your Airflow deployment.

tests/system/google/cloud/bigquery/example_bigquery_queries_async.py

insert_query_job = BigQueryInsertJobOperator(
    task_id="insert_query_job",
    configuration={
        "query": {
            "query": INSERT_ROWS_QUERY,
            "useLegacySql": False,
            "priority": "BATCH",
        }
    },
    location=LOCATION,
    deferrable=True,
)

For more information on types of BigQuery job please check documentation.

If you want to include some files in your configuration you can use include clause of Jinja template language as follow:

tests/system/google/cloud/bigquery/example_bigquery_queries.py

select_query_job = BigQueryInsertJobOperator(
    task_id="select_query_job",
    configuration={
        "query": {
            "query": "{% include QUERY_SQL_PATH %}",
            "useLegacySql": False,
        }
    },
)

The included file can also use Jinja templates which can be useful in case of .sql files.

Additionally you can use job_id parameter of BigQueryInsertJobOperator to improve idempotency. If this parameter is not passed then uuid will be used as job_id. If provided then operator will try to submit a new job with this job_id`. If there’s already a job with such job_id then it will reattach to the existing job.

Also for all this action you can use operator in the deferrable mode:

tests/system/google/cloud/bigquery/example_bigquery_queries_async.py

insert_query_job = BigQueryInsertJobOperator(
    task_id="insert_query_job",
    configuration={
        "query": {
            "query": INSERT_ROWS_QUERY,
            "useLegacySql": False,
            "priority": "BATCH",
        }
    },
    location=LOCATION,
    deferrable=True,
)

Durable execution¶

BigQueryInsertJobOperator submits a job and then polls it to completion on the worker. By default the operator runs in a durable mode that makes this crash-safe: the submitted BigQuery job id is persisted to task state store before polling begins, so if the worker crashes or is preempted and the task is retried, the operator reconnects to the job that is already running in BigQuery instead of submitting a duplicate.

On retry the operator checks the prior job’s state:

if it is still running, the operator reconnects and continues polling
if it already succeeded, the operator returns immediately without resubmitting
if it failed, or its id is no longer found, the operator submits the job fresh

This makes reattachment on retry work regardless of force_rerun, which defaults to True. Without durable execution, force_rerun=True computes a new random job id on every attempt, so the existing reattach_states/Conflict mechanism (recomputing an identical id and relying on BigQuery rejecting the duplicate) almost never has a matching id to find on retry – setting force_rerun=False was the only way to make that mechanism reliable. Durable execution replaces that specific use of force_rerun=False: the operator now reconnects using the id it actually persisted, so force_rerun no longer needs to change purely to make retries of the same task instance reattach.

force_rerun still matters for a different, narrower purpose it always had: idempotency across separate invocations, such as two different DAG runs or a manual re-trigger that reuses the same explicit job_id. Durable execution’s persisted state is scoped to a single task instance (dag_id/task_id/run_id/map_index); a new DAG run has no durable state to reconnect to at all. If you rely on force_rerun=False with an explicit job_id so that re-triggering the same logical run doesn’t submit a second job, that guarantee is unrelated to durable execution and keeping force_rerun=False is still the right choice for it.

Durable execution requires Airflow 3.3 or newer, since it relies on the task state store. On earlier Airflow versions the flag is a no-op and the operator always submits a fresh job on retry, exactly as before – including the pre-existing reattach_states/Conflict behavior, which is unchanged. If the task state store is unavailable at runtime, the operator logs that crash recovery is disabled and behaves the same way.

To opt out and always submit a fresh job on retry, set durable=False:

insert_query_job = BigQueryInsertJobOperator(
    task_id="insert_query_job",
    configuration={
        "query": {
            "query": "SELECT 1",
            "useLegacySql": False,
        }
    },
    durable=False,
)

Durable execution applies to the synchronous path. When deferrable=True is set, the Triggerer already tracks the job across the wait, so deferrable mode takes precedence and durable has no effect.

Validate data¶

Check if query result has data¶

To perform checks against BigQuery you can use BigQueryCheckOperator

This operator expects a sql query that will return a single row. Each value on that first row is evaluated using python bool casting. If any of the values return False the check is failed and errors out.

tests/system/google/cloud/bigquery/example_bigquery_queries.py

check_count = BigQueryCheckOperator(
    task_id="check_count",
    sql=f"SELECT COUNT(*) FROM {DATASET_NAME}.{TABLE_1}",
    use_legacy_sql=False,
)

Also you can use deferrable mode in this operator

tests/system/google/cloud/bigquery/example_bigquery_queries_async.py

check_count = BigQueryCheckOperator(
    task_id="check_count",
    sql=f"SELECT COUNT(*) FROM {DATASET_NAME}.{TABLE_NAME_1}",
    use_legacy_sql=False,
    location=LOCATION,
    deferrable=True,
)

Compare query result to pass value¶

To perform a simple value check using sql code you can use BigQueryValueCheckOperator

These operators expects a sql query that will return a single row. Each value on that first row is evaluated against pass_value which can be either a string or numeric value. If numeric, you can also specify tolerance.

tests/system/google/cloud/bigquery/example_bigquery_queries.py

check_value = BigQueryValueCheckOperator(
    task_id="check_value",
    sql=f"SELECT COUNT(*) FROM {DATASET_NAME}.{TABLE_1}",
    pass_value=4,
    use_legacy_sql=False,
)

Also you can use deferrable mode in this operator

tests/system/google/cloud/bigquery/example_bigquery_queries_async.py

check_value = BigQueryValueCheckOperator(
    task_id="check_value",
    sql=f"SELECT COUNT(*) FROM {DATASET_NAME}.{TABLE_NAME_1}",
    pass_value=2,
    use_legacy_sql=False,
    location=LOCATION,
    deferrable=True,
)

Compare metrics over time¶

To check that the values of metrics given as SQL expressions are within a certain tolerance of the ones from days_back before you can either use BigQueryIntervalCheckOperator or BigQueryIntervalCheckAsyncOperator

tests/system/google/cloud/bigquery/example_bigquery_queries.py

check_interval = BigQueryIntervalCheckOperator(
    task_id="check_interval",
    table=f"{DATASET_NAME}.{TABLE_1}",
    days_back=1,
    metrics_thresholds={"COUNT(*)": 1.5},
    use_legacy_sql=False,
)

Also you can use deferrable mode in this operator

tests/system/google/cloud/bigquery/example_bigquery_queries_async.py

check_interval = BigQueryIntervalCheckOperator(
    task_id="check_interval",
    table=f"{DATASET_NAME}.{TABLE_NAME_1}",
    days_back=1,
    metrics_thresholds={"COUNT(*)": 1.5},
    use_legacy_sql=False,
    location=LOCATION,
    deferrable=True,
)

Check columns with predefined tests¶

To check that columns pass user-configurable tests you can use BigQueryColumnCheckOperator

tests/system/google/cloud/bigquery/example_bigquery_queries.py

column_check = BigQueryColumnCheckOperator(
    task_id="column_check",
    table=f"{DATASET_NAME}.{TABLE_1}",
    column_mapping={"value": {"null_check": {"equal_to": 0}}},
    use_legacy_sql=False,
)

Check table level data quality¶

To check that tables pass user-defined tests you can use BigQueryTableCheckOperator

tests/system/google/cloud/bigquery/example_bigquery_queries.py

table_check = BigQueryTableCheckOperator(
    task_id="table_check",
    table=f"{DATASET_NAME}.{TABLE_1}",
    checks={"row_count_check": {"check_statement": "COUNT(*) = 4"}},
    use_legacy_sql=False,
)

Sensors¶

Check that a Table exists¶

To check that a table exists you can define a sensor operator. This allows delaying execution of downstream operators until a table exist. If the table is sharded on dates you can for instance use the {{ ds_nodash }} macro as the table name suffix.

BigQueryTableExistenceSensor.

tests/system/google/cloud/bigquery/example_bigquery_sensors.py[source]

check_table_exists = BigQueryTableExistenceSensor(
    task_id="check_table_exists", project_id=PROJECT_ID, dataset_id=DATASET_NAME, table_id=TABLE_NAME
)

Also you can use deferrable mode in this operator if you would like to free up the worker slots while the sensor is running.

tests/system/google/cloud/bigquery/example_bigquery_sensors.py[source]

check_table_exists_def = BigQueryTableExistenceSensor(
    task_id="check_table_exists_def",
    project_id=PROJECT_ID,
    dataset_id=DATASET_NAME,
    table_id=TABLE_NAME,
    deferrable=True,
)

tests/system/google/cloud/bigquery/example_bigquery_sensors.py[source]

check_table_exists_async = BigQueryTableExistenceSensor(
    task_id="check_table_exists_async",
    project_id=PROJECT_ID,
    dataset_id=DATASET_NAME,
    table_id=TABLE_NAME,
)

Check that a Table Partition exists¶

To check that a table exists and has a partition you can use. BigQueryTablePartitionExistenceSensor.

tests/system/google/cloud/bigquery/example_bigquery_sensors.py[source]

check_table_partition_exists = BigQueryTablePartitionExistenceSensor(
    task_id="check_table_partition_exists",
    project_id=PROJECT_ID,
    dataset_id=DATASET_NAME,
    table_id=TABLE_NAME,
    partition_id=PARTITION_NAME,
)

For DAY partitioned tables, the partition_id parameter is a string on the “%Y%m%d” format

Also you can use deferrable mode in this operator if you would like to free up the worker slots while the sensor is running.

tests/system/google/cloud/bigquery/example_bigquery_sensors.py[source]

check_table_partition_exists_def = BigQueryTablePartitionExistenceSensor(
    task_id="check_table_partition_exists_def",
    project_id=PROJECT_ID,
    dataset_id=DATASET_NAME,
    table_id=TABLE_NAME,
    partition_id=PARTITION_NAME,
    deferrable=True,
)

tests/system/google/cloud/bigquery/example_bigquery_sensors.py[source]

check_table_partition_exists_async = BigQueryTablePartitionExistenceSensor(
    task_id="check_table_partition_exists_async",
    partition_id=PARTITION_NAME,
    project_id=PROJECT_ID,
    dataset_id=DATASET_NAME,
    table_id=TABLE_NAME,
)

Check that the BigQuery Table Streaming Buffer is empty¶

To check that the BigQuery streaming buffer of a table is empty you can use BigQueryStreamingBufferEmptySensor. This sensor is useful in ETL pipelines to ensure that recent streamed data has been fully processed before continuing downstream tasks.

tests/system/google/cloud/bigquery/example_bigquery_streaming_buffer_sensor.py[source]

check_streaming_buffer_empty = BigQueryStreamingBufferEmptySensor(
    task_id="check_streaming_buffer_empty",
    project_id=PROJECT_ID,
    dataset_id=DATASET_NAME,
    table_id=TABLE_NAME,
    poke_interval=30,
    timeout=5400,  # BigQuery flushes the streaming buffer within ~90 minutes
)

Also you can use deferrable mode in this operator if you would like to free up the worker slots while the sensor is running.

tests/system/google/cloud/bigquery/example_bigquery_streaming_buffer_sensor.py[source]

check_streaming_buffer_empty_def = BigQueryStreamingBufferEmptySensor(
    task_id="check_streaming_buffer_empty_def",
    project_id=PROJECT_ID,
    dataset_id=DATASET_NAME,
    table_id=TABLE_NAME,
    deferrable=True,
    poke_interval=30,
    timeout=5400,  # BigQuery flushes the streaming buffer within ~90 minutes
)

Reference¶

For further information, look at: