How to Call BigQuery Stored Procedure from Cloud Composer?

Automating your data warehouse workflows in Google Cloud is critical for scaling your analytics. If you want to orchestrate complex SQL logic seamlessly, knowing how to call a BigQuery stored procedure from Cloud Composer is an essential skill for any data engineer. Here is a complete, step-by-step guide and Airflow DAG example to get your pipelines running smoothly.

Page Content

• Introduction
• Step 1: Create the BigQuery Stored Procedure
• Step 2: Write the Cloud Composer (Airflow) DAG
• Best Practices for Orchestrating BigQuery
• Conclusion

Introduction

Google Cloud offers several ways to orchestrate workloads, but Cloud Composer (which is built on managed Apache Airflow) remains the top choice for complex, multi-step data pipelines.

While BigQuery is fantastic at processing massive datasets, combining it with Airflow allows you to automate everything from daily reporting to dynamic ETL jobs. By wrapping your complex SQL transformations inside a BigQuery stored procedure, you keep your code clean and let the database handle the heavy lifting.

In this guide, we will walk through exactly how to create a stored procedure in BigQuery and how to trigger it using a standard Cloud Composer DAG.

Step 1: Create the BigQuery Stored Procedure

Before we can orchestrate anything in Airflow, we need a stored procedure to call. Stored procedures in BigQuery allow you to run multiple SQL statements, declare variables, and implement control flow (like IF statements and loops) directly within the database.

Here is a practical example. We will create a procedure that generates a unique ID, inserts it into a customer table, and returns a confirmation string.

CREATE OR REPLACE PROCEDURE `team.ds_tbl_db.create_customer`()
BEGIN
  -- Declare a variable to hold the new UUID
  DECLARE id STRING;
  SET id = GENERATE_UUID();
  
  -- Insert the new ID into our customer table
  INSERT INTO `team.ds_tbl_db.customers` (customer_id) 
  VALUES(id);
  
  -- Return a formatted success message
  SELECT FORMAT("Created customer %s", id);
END;

Testing Your Procedure: You can test this directly in the BigQuery console by running the CALL statement, and then querying your table to ensure the data was inserted:

CALL `team.ds_tbl_db.create_customer`();
SELECT * FROM `team.ds_tbl_db.customers`;

Step 2: Write the Cloud Composer (Airflow) DAG

Now that our procedure is ready, we need to orchestrate it using Cloud Composer.

To do this, we will use the BigQueryOperator in Apache Airflow. This operator submits the CALL statement to BigQuery exactly as if you were typing it into the console yourself.

Here is the complete Python DAG code to schedule and run our stored procedure daily:

import datetime
import airflow
from airflow.contrib.operators import bigquery_operator

# Set a dynamic start date for the DAG
YESTERDAY = datetime.datetime.now() - datetime.timedelta(days=1)

# Define the default arguments for our tasks
default_args = {
    'owner': 'BigQuery SP Example',
    'depends_on_past': False,
    'email': ['your-email@example.com'],
    'email_on_failure': False,
    'email_on_retry': False,
    'retries': 1,
    'retry_delay': datetime.timedelta(minutes=5),
    'start_date': YESTERDAY,
}

# Initialize the DAG
with airflow.DAG(
    'call_stored_procedure_Sample', 
    catchup=False, 
    default_args=default_args, 
    schedule_interval=datetime.timedelta(days=1)
) as dag:

    # Define the BigQuery Operator task
    call_stored_procedure = bigquery_operator.BigQueryOperator(
        task_id="call_stored_procedure",
        sql="""CALL `team.ds_tbl_db.create_customer`();""",
        use_legacy_sql=False,
    )

Best Practices for Orchestrating BigQuery

When you call a BigQuery stored procedure from Cloud Composer, keep these quick tips in mind:

• Always use Standard SQL: BigQuery legacy SQL does not support stored procedures. Ensure use_legacy_sql=False is explicitly set in your Airflow operator.
• Keep Airflow Light: Do not pull massive amounts of BigQuery data into Airflow memory. Let the stored procedure do the data transformations inside BigQuery, and use Airflow purely to trigger the job and manage dependencies.
• Update Your Operators (Composer 2.x): If you are running a newer version of Cloud Composer (Airflow 2.0+), it is recommended to transition from airflow.contrib.operators to the updated BigQueryInsertJobOperator found in the modern Google Cloud provider packages.

Conclusion

By combining the processing power of BigQuery stored procedures with the scheduling reliability of Cloud Composer, you can build incredibly robust data pipelines. Stored procedures keep your SQL logic centralized in the data warehouse, while Airflow ensures your jobs run exactly when they are supposed to.