GCP operators (BigQuery, GCS, Dataflow) in Apache Airflow - Commands & Configuration

Introduction

When you want to automate tasks that use Google Cloud services like BigQuery, Cloud Storage, or Dataflow, you can use GCP operators in Airflow. These operators help you run jobs and move data without writing complex code for each service.

When you need to load data from Cloud Storage into BigQuery automatically every day.

When you want to run a Dataflow job to process streaming data as part of your workflow.

When you want to move files between different Cloud Storage buckets on a schedule.

When you want to create a BigQuery table or run a query as part of a data pipeline.

When you want to chain multiple Google Cloud tasks together in a reliable order.

Config File - gcp_data_pipeline_dag.py

gcp_data_pipeline_dag.py

from airflow import DAG
from airflow.utils.dates import days_ago
from airflow.providers.google.cloud.operators.bigquery import BigQueryInsertJobOperator
from airflow.providers.google.cloud.operators.gcs import GCSListObjectsOperator
from airflow.providers.google.cloud.operators.dataflow import DataflowStartFlexTemplateOperator
from datetime import timedelta

default_args = {
    'owner': 'airflow',
    'start_date': days_ago(1),
    'retries': 1,
    'retry_delay': timedelta(minutes=5),
}

dag = DAG(
    'gcp_data_pipeline',
    default_args=default_args,
    schedule_interval='@daily',
    catchup=False
)

list_gcs_files = GCSListObjectsOperator(
    task_id='list_gcs_files',
    bucket='my-data-bucket',
    prefix='input/',
    dag=dag
)

run_bigquery_query = BigQueryInsertJobOperator(
    task_id='run_bigquery_query',
    configuration={
        'query': {
            'query': 'SELECT * FROM `my-project.my_dataset.my_table` LIMIT 10',
            'useLegacySql': False
        }
    },
    location='US',
    dag=dag
)

start_dataflow_job = DataflowStartFlexTemplateOperator(
    task_id='start_dataflow_job',
    body={
        'launchParameter': {
            'jobName': 'example-dataflow-job',
            'containerSpecGcsPath': 'gs://my-templates/dataflow-template.json',
            'parameters': {
                'inputFile': 'gs://my-data-bucket/input/data.csv',
                'outputTable': 'my-project:my_dataset.output_table'
            }
        }
    },
    location='us-central1',
    dag=dag
)

list_gcs_files >> run_bigquery_query >> start_dataflow_job

This Airflow DAG defines a simple data pipeline using GCP operators.

• GCSListObjectsOperator lists files in a Cloud Storage bucket.
• BigQueryInsertJobOperator runs a SQL query on BigQuery.
• DataflowStartFlexTemplateOperator starts a Dataflow job from a Flex Template.
• Tasks are linked to run in order: list files, then query BigQuery, then start Dataflow.

Commands

List all available DAGs to confirm the new GCP data pipeline DAG is recognized by Airflow.

Terminal

airflow dags list

Expected Output

gcp_data_pipeline

example_dag

Manually trigger the GCP data pipeline DAG to start the workflow immediately.

Terminal

airflow dags trigger gcp_data_pipeline

Expected Output

Created <DagRun gcp_data_pipeline @ 2024-06-01T12:00:00+00:00: manual__2024-06-01T12:00:00+00:00, externally triggered: True>

List all tasks in the gcp_data_pipeline DAG to see the steps involved.

Terminal

airflow tasks list gcp_data_pipeline

Expected Output

list_gcs_files

run_bigquery_query

start_dataflow_job

Test the list_gcs_files task for the given date to verify it lists files from the GCS bucket correctly.

Terminal

airflow tasks test gcp_data_pipeline list_gcs_files 2024-06-01

Expected Output

[2024-06-01 12:00:00,000] {taskinstance.py:XXXX} INFO - Starting task: list_gcs_files

[2024-06-01 12:00:01,000] {gcs.py:XXXX} INFO - Found files: ['input/data.csv', 'input/data2.csv']

[2024-06-01 12:00:02,000] {taskinstance.py:XXXX} INFO - Task list_gcs_files succeeded

Key Concept

If you remember nothing else from this pattern, remember: GCP operators in Airflow let you automate Google Cloud tasks easily by wrapping complex API calls into simple tasks.

Common Mistakes

Not setting the correct Google Cloud connection in Airflow before running GCP operators.

The operators need credentials to access GCP services; without a proper connection, tasks will fail with authentication errors.

Create and configure a Google Cloud connection in Airflow UI or via CLI with the right service account credentials before running the DAG.

Using legacy SQL syntax in BigQuery queries without setting 'useLegacySql' to False.

BigQuery defaults to legacy SQL which may cause query errors if standard SQL syntax is used.

Always set 'useLegacySql': False in the BigQueryInsertJobOperator configuration when using standard SQL.

Not specifying the correct location for Dataflow jobs or BigQuery queries.

GCP services require location to route requests; missing or wrong location causes job failures.

Set the 'location' parameter correctly matching your GCP resources region.

Summary

Use GCP operators in Airflow to automate tasks with BigQuery, Cloud Storage, and Dataflow.

Define a DAG with tasks that list GCS files, run BigQuery queries, and start Dataflow jobs.

Trigger and test the DAG using Airflow CLI commands to verify the workflow.

Practice

1. What is the main purpose of GCP operators in Airflow when working with BigQuery, GCS, and Dataflow?

easy

A. To create virtual machines in Google Cloud

B. To manually write SQL queries for BigQuery

C. To replace Google Cloud Console for all cloud services

D. To automate and manage cloud tasks like queries, storage, and data processing

Solution

1. Step 1: Understand the role of GCP operators in Airflow

   GCP operators are designed to automate cloud tasks such as running queries in BigQuery, managing files in GCS, and launching Dataflow jobs.

2. Step 2: Identify the correct purpose among options

   Only To automate and manage cloud tasks like queries, storage, and data processing correctly describes automation and management of cloud tasks, while others describe unrelated or manual tasks.

3. Final Answer:

   To automate and manage cloud tasks like queries, storage, and data processing -> Option D

4. Quick Check:

   GCP operators automate cloud tasks = A [OK]

Hint: GCP operators automate cloud tasks, not manual or unrelated actions [OK]

Common Mistakes:

• Thinking GCP operators replace all Google Cloud Console functions
• Confusing manual SQL writing with automation
• Assuming GCP operators create virtual machines

2. Which of the following is the correct way to import the BigQuery operator in an Airflow DAG?

easy

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

B. import airflow.gcp.bigquery as bq

C. from airflow.operators.bigquery import BigQueryOperator

D. from google.cloud import bigquery

Solution

1. Step 1: Recall the correct import path for BigQuery operators in Airflow

   Airflow uses provider packages; the BigQuery operator is imported from airflow.providers.google.cloud.operators.bigquery.

2. Step 2: Match the correct operator class and import syntax

   from airflow.providers.google.cloud.operators.bigquery import BigQueryInsertJobOperator uses the correct module and operator class name. Options B and C use incorrect paths or outdated names. from google.cloud import bigquery imports the Google Cloud client library, not Airflow operator.

3. Final Answer:

   from airflow.providers.google.cloud.operators.bigquery import BigQueryInsertJobOperator -> Option A

4. Quick Check:

   Correct Airflow BigQuery import = D [OK]

Hint: Airflow GCP operators import from airflow.providers.google.cloud.operators [OK]

Common Mistakes:

• Using outdated or incorrect import paths
• Confusing Google Cloud client libraries with Airflow operators
• Importing non-existent modules

3. Given this Airflow task using BigQueryInsertJobOperator:

bigquery_task = BigQueryInsertJobOperator(
    task_id='run_query',
    configuration={
        'query': {
            'query': 'SELECT COUNT(*) FROM `my_dataset.my_table`',
            'useLegacySql': False
        }
    }
)

What will this task do when executed?

medium

A. Upload data from GCS to BigQuery

B. Create a new BigQuery table named my_table

C. Run a BigQuery SQL query that counts rows in my_dataset.my_table

D. Delete the table my_dataset.my_table

Solution

1. Step 1: Analyze the BigQueryInsertJobOperator configuration

   The configuration specifies a SQL query: 'SELECT COUNT(*) FROM `my_dataset.my_table`' with standard SQL enabled (useLegacySql: False).

2. Step 2: Determine the task action from the query

   The query counts rows in the specified table. The operator runs this query as a job in BigQuery.

3. Final Answer:

   Run a BigQuery SQL query that counts rows in my_dataset.my_table -> Option C

4. Quick Check:

   BigQueryInsertJobOperator runs SQL queries = A [OK]

Hint: Look at the 'query' field to see what SQL runs [OK]

Common Mistakes:

• Assuming it creates or deletes tables
• Confusing query execution with data upload
• Ignoring useLegacySql flag

4. You wrote this Airflow task to upload a file to GCS:

upload_task = GCSToBigQueryOperator(
    task_id='upload_to_gcs',
    bucket='my_bucket',
    source_objects=['data.csv'],
    destination_project_dataset_table='my_dataset.my_table'
)

What is wrong with this task?

medium

A. GCSToBigQueryOperator is used incorrectly to upload to GCS; it loads data from GCS to BigQuery

B. The bucket name must be omitted for this operator

C. source_objects should be a string, not a list

D. destination_project_dataset_table should be a GCS path

Solution

1. Step 1: Understand GCSToBigQueryOperator purpose

   This operator loads data from Google Cloud Storage (GCS) into BigQuery tables; it does not upload files to GCS.

2. Step 2: Identify the misuse in the task

   The task tries to upload a file to GCS using GCSToBigQueryOperator, which is incorrect. Uploading to GCS requires a different operator like GCSUploadOperator.

3. Final Answer:

   GCSToBigQueryOperator is used incorrectly to upload to GCS; it loads data from GCS to BigQuery -> Option A

4. Quick Check:

   GCSToBigQueryOperator loads data from GCS, not upload to GCS [OK]

Hint: Check operator purpose: GCSToBigQuery loads data, doesn't upload files [OK]

Common Mistakes:

• Using GCSToBigQueryOperator to upload files to GCS
• Confusing source_objects type
• Misunderstanding destination table format

5. You want to run a Dataflow job in Airflow that reads data from GCS, processes it, and writes results back to BigQuery. Which combination of operators is best to build this workflow?

hard

A. Use DataflowCreatePythonJobOperator to run the Dataflow job, and BigQueryInsertJobOperator to run SQL queries

B. Use DataflowCreatePythonJobOperator to run Dataflow job that reads/writes data, no separate BigQuery operator needed

C. Use DataflowCreateJavaJobOperator to run Dataflow, and GCSToBigQueryOperator to load data from GCS to BigQuery

D. Use GCSCreateBucketOperator to create buckets, then DataflowTemplatedJobStartOperator to run Dataflow

Solution

1. Step 1: Understand Dataflow job capabilities

   Dataflow jobs can read from GCS, process data, and write directly to BigQuery within the same job.

2. Step 2: Choose the operator that runs the Dataflow job end-to-end

   DataflowCreatePythonJobOperator runs Python Dataflow jobs that handle reading and writing data, so no separate BigQuery operator is needed for writing results.

3. Step 3: Evaluate other options

   Use DataflowCreatePythonJobOperator to run the Dataflow job, and BigQueryInsertJobOperator to run SQL queries runs Dataflow and BigQuery separately, which is unnecessary if Dataflow handles both. Use GCSCreateBucketOperator to create buckets, then DataflowTemplatedJobStartOperator to run Dataflow involves bucket creation, not relevant here. Use DataflowCreateJavaJobOperator to run Dataflow, and GCSToBigQueryOperator to load data from GCS to BigQuery uses Java job and GCSToBigQueryOperator, which is redundant.

4. Final Answer:

   Use DataflowCreatePythonJobOperator to run Dataflow job that reads/writes data, no separate BigQuery operator needed -> Option B

5. Quick Check:

   Dataflow job can read/write data itself = B [OK]

Hint: Dataflow jobs can handle GCS to BigQuery in one step [OK]

Common Mistakes:

• Running separate BigQuery operators unnecessarily
• Confusing Dataflow Java and Python operators
• Adding unrelated bucket creation tasks