+1 (415) 943-1448
BQBrains

PySpark Inside BigQuery: Stored Procedures for Apache Spark and Studio Notebooks

Most BigQuery work should stay in SQL. But every estate eventually hits jobs that SQL handles badly: parsing a directory of gnarly fixed-width files, calling a Python library that has no SQL equivalent, iterating over thousands of small partitions, or reusing a PySpark transformation a team already owns. For those, you do not need a separate Dataproc cluster with its own IAM model and its own on-call. BigQuery runs serverless Apache Spark from inside the warehouse, in two forms: stored procedures for Apache Spark, callable from SQL, and PySpark notebooks in BigQuery Studio.

This tutorial sets both up end to end, then gives you a decision rule for when to reach for Spark at all.

What you get, and what it costs

A stored procedure for Apache Spark is a SQL-callable object whose body is PySpark (or Java/Scala) code. When you CALL it, BigQuery provisions a serverless Spark runtime, runs your code with the BigQuery Spark connector already available, and tears the runtime down. There is no cluster to size, patch, or forget about.

Two things to know before you start:

  • Spark jobs are not billed as BigQuery slots or bytes scanned. They bill under Google Cloud Serverless for Apache Spark (DCU-hours plus shuffle storage), on a separate line of your bill. A reservation will not cap them — set project quotas and budgets instead.
  • Cold start is real. Expect tens of seconds before your first line of Python executes. Spark procedures are for batch work, never for anything a dashboard waits on.

Step 1: enable the APIs and create a connection

Spark procedures run under a connection of type SPARK, which owns a service account. Create it once per region.

gcloud services enable bigqueryconnection.googleapis.com \
  dataproc.googleapis.com \
  bigquery.googleapis.com

bq mk --connection \
  --connection_type=SPARK \
  --location=US \
  spark-conn

Find the service account the connection created, then grant it only what the job needs:

bq show --connection --location=US acme-analytics.US.spark-conn
# note spark.serviceAccountId, e.g. bqcx-123456789-ab12@gcp-sa-bigquery-consp.iam.gserviceaccount.com

SA=bqcx-123456789-ab12@gcp-sa-bigquery-consp.iam.gserviceaccount.com

gcloud projects add-iam-policy-binding acme-analytics \
  --member="serviceAccount:$SA" --role="roles/bigquery.dataEditor"
gcloud projects add-iam-policy-binding acme-analytics \
  --member="serviceAccount:$SA" --role="roles/bigquery.jobUser"
gsutil iam ch "serviceAccount:$SA:objectViewer" gs://acme-raw-landing

Resist the urge to hand it roles/editor. The connection service account is a standing identity that any user with permission to CALL the procedure can borrow — scope it to the datasets and buckets the job actually touches. The caller also needs roles/bigquery.connectionUser on the connection.

Step 2: write your first Spark stored procedure

The inline form keeps the Python next to the SQL, which is fine for short jobs:

CREATE OR REPLACE PROCEDURE `acme-analytics.ops.parse_edi_drops`()
WITH CONNECTION `acme-analytics.US.spark-conn`
OPTIONS (engine = 'SPARK', runtime_version = '2.2')
LANGUAGE python AS r"""
from pyspark.sql import SparkSession, functions as F

spark = SparkSession.builder.appName("parse_edi_drops").getOrCreate()

# Fixed-width EDI drops that SQL string functions handle badly
raw = spark.read.text("gs://acme-raw-landing/edi/2026/*/*.txt")

parsed = (
    raw
    .withColumn("record_type", F.trim(F.substring("value", 1, 3)))
    .withColumn("partner_id",  F.trim(F.substring("value", 4, 12)))
    .withColumn("order_ts",    F.to_timestamp(F.substring("value", 16, 14), "yyyyMMddHHmmss"))
    .withColumn("amount",      (F.substring("value", 30, 11).cast("double") / 100))
    .filter(F.col("record_type") == "ORD")
    .drop("value")
)

(
    parsed.write
    .format("bigquery")
    .option("table", "acme-analytics.staging.edi_orders")
    .option("writeMethod", "direct")          # Storage Write API, no GCS staging bucket
    .option("partitionField", "order_ts")
    .option("partitionType", "DAY")
    .mode("overwrite")
    .save()
)
"""
;

CALL `acme-analytics.ops.parse_edi_drops`();

Notes that save debugging time:

  • Use r""" so backslashes in regexes survive the SQL parser.
  • writeMethod = "direct" uses the Storage Write API and skips the temporary GCS bucket entirely. The older indirect method needs temporaryGcsBucket set.
  • runtime_version pins the Spark and Python versions. Pin it explicitly; do not inherit whatever the default becomes next quarter.

For anything longer than a screen, store the code in GCS and reference it, so it can live in Git and go through code review:

CREATE OR REPLACE PROCEDURE `acme-analytics.ops.parse_edi_drops`(run_date DATE)
WITH CONNECTION `acme-analytics.US.spark-conn`
OPTIONS (
  engine = 'SPARK',
  runtime_version = '2.2',
  main_file_uri = 'gs://acme-code/spark/parse_edi_drops.py',
  properties = [("spark.executor.instances", "4")]
);

Step 3: read from BigQuery, not from GCS exports

The Spark connector pushes projections and filters down to the BigQuery Storage Read API, so read tables directly rather than exporting them first:

orders = (
    spark.read.format("bigquery")
    .option("table", "acme-analytics.warehouse.fct_orders")
    .option("filter", "order_date >= '2026-01-01'")   # pushed down, not read then dropped
    .load()
    .select("order_id", "customer_id", "order_date", "net_amount")
)

Two habits matter for cost: always select() the columns you need (the connector reads only those), and always pass filter for partition pruning. A Spark job that reads a full 40 TB fact table because someone wrote .load() and filtered in Python is the single most common way these jobs get expensive.

Step 4: passing arguments and returning results

Procedure arguments arrive as Spark properties, which keeps parameterised backfills readable:

CREATE OR REPLACE PROCEDURE `acme-analytics.ops.rebuild_features`(run_date DATE)
WITH CONNECTION `acme-analytics.US.spark-conn`
OPTIONS (engine = 'SPARK', runtime_version = '2.2')
LANGUAGE python AS r"""
from pyspark.sql import SparkSession
spark = SparkSession.builder.getOrCreate()
run_date = spark.conf.get("spark.sql.legacy.bigquery.procedure.arg.run_date")
...
"""
;

Spark procedures do not return result sets. The pattern that works is: write to a table, then let the calling SQL script read it.

BEGIN
  CALL `acme-analytics.ops.rebuild_features`(CURRENT_DATE());
  MERGE `acme-analytics.warehouse.dim_customer_features` t
  USING `acme-analytics.staging.customer_features` s
  ON t.customer_id = s.customer_id
  WHEN MATCHED THEN UPDATE SET t.features = s.features, t.updated_at = CURRENT_TIMESTAMP()
  WHEN NOT MATCHED THEN INSERT ROW;
END;

Wrap that script in a scheduled query or a Dataform operation and the Spark step becomes an ordinary node in your ELT graph — no second orchestrator.

Step 5: PySpark notebooks in BigQuery Studio

For exploration rather than production, BigQuery Studio notebooks (Colab Enterprise runtimes) speak both SQL and PySpark in the same file. Create a notebook, attach a Spark session backed by the same connection, and you can prototype the exact code you will later paste into a procedure. The workflow that keeps notebooks from becoming shadow pipelines:

  1. Explore in the notebook, on a LIMITed or date-scoped slice.
  2. Move the working transformation into a .py file in Git.
  3. Deploy it as a stored procedure with main_file_uri, called by Dataform or a scheduled query.
  4. Keep the notebook as documentation, not as the production artifact.

Step 6: observability

Spark jobs do not show in INFORMATION_SCHEMA.JOBS as query jobs with slot metrics, so build the habit of checking both places:

SELECT job_id, creation_time, statement_type, query, state, error_result
FROM `region-us`.INFORMATION_SCHEMA.JOBS_BY_PROJECT
WHERE creation_time > TIMESTAMP_SUB(CURRENT_TIMESTAMP(), INTERVAL 1 DAY)
  AND statement_type = 'CALL'
ORDER BY creation_time DESC;

For the Spark side — executor logs, the Spark History Server UI, shuffle metrics — follow the batch link on the job in the BigQuery console, which opens the corresponding Serverless for Apache Spark batch. Log failures to Cloud Logging with a log-based metric on the procedure name so a silent Spark failure does not leave yesterday's table in place while every downstream dashboard looks healthy.

When not to use Spark

Be honest about this list, because Spark inside BigQuery is easy to overuse:

  • Joins, aggregations, window functions, MERGE. SQL wins on cost, speed, and maintainability. Always.
  • Python UDF-shaped logic on modest data. Use a remote function on Cloud Run or BigQuery DataFrames instead.
  • pandas-style dataframe work. BigQuery DataFrames pushes pandas down to BigQuery with no Spark runtime at all.
  • Anything latency-sensitive. Cold start alone disqualifies it.

Reach for Spark when you have unstructured or awkward file formats, an existing PySpark codebase worth reusing, a Python library with no SQL equivalent, or genuinely iterative algorithms. Those cases are real — they are just rarer than the number of Spark jobs in the average warehouse suggests.

Checklist

  • One SPARK connection per region, with a narrowly scoped service account.
  • runtime_version pinned explicitly in every procedure.
  • Production code in GCS via main_file_uri, versioned in Git — not inline.
  • writeMethod = "direct" for writes; select() and filter pushdown for reads.
  • Spark steps called from a SQL script or Dataform so lineage stays in one graph.
  • Alerting on CALL failures, not just on the SQL around them.

If you are weighing whether a workload belongs in SQL, in Spark, or in a pipeline you should not be running at all, our data engineering team does this triage for a living — get in touch and we will look at the job list with you.