Great question. In Prefect 3.x, the best practice is to inject credentials (not client objects) into your tasks/flows and create the BigQuery client inside the task using those credentials. This keeps things serializable/picklable and works with all task runners, including process-based ones. Here are the recommended patterns: 1) Pass GcpCredentials to tasks and build the client inside - Store your GCP service account as a Prefect block (

GcpCredentials

) - Load the block in the flow and pass it to tasks - Inside the task, call

get_bigquery_client(...)

from prefect import flow, task
from prefect_gcp.credentials import GcpCredentials

@task
def run_query(query: str, creds: GcpCredentials, project: str | None = None, location: str | None = None):
    client = creds.get_bigquery_client(project=project, location=location)
    return list(client.query(query).result())

@flow
def pipeline():
    creds = GcpCredentials.load("my-gcp-credentials")  # created/saved once in Prefect
    rows = run_query("SELECT 1 AS x", creds, project="my-project", location="US")
    return rows

Why this is preferred - BigQuery client objects are not picklable; passing them between processes breaks with

ProcessPoolTaskRunner

-

GcpCredentials

blocks are serializable and safe to pass as parameters - Each task can build and use its own client quickly and safely 2) Use the prefect-gcp BigQuery helper tasks - If you don’t need full client control, the integration tasks handle auth and common operations

from prefect import flow
from prefect_gcp.credentials import GcpCredentials
from prefect_gcp.bigquery import bigquery_query

@flow
def run():
    creds = GcpCredentials.load("my-gcp-credentials")
    df = bigquery_query(
        query="SELECT * FROM `project.dataset.table` LIMIT 10",
        gcp_credentials=creds,
        to_dataframe=True,
    )
    return df

3) Notes and gotchas - Don’t pass actual BigQuery Client objects to tasks if you might use process-based runners or remote workers (Kubernetes, Docker). Create the client inside the task from credentials. - Loading the block inside the task is okay for sequential/threaded execution, but for process-based runners, it’s more robust to load in the flow and pass the block to tasks. - You can use Application Default Credentials (ADC) by creating a

GcpCredentials

block without a service account and ensuring auth is available in the environment (e.g.,

GOOGLE_APPLICATION_CREDENTIALS

or

gcloud auth application-default login

). - Ensure your runtime has

google-cloud-bigquery

and

prefect-gcp

installed. Helpful links - Prefect GCP integration: prefect-gcp docs - BigQuery helpers: prefect-gcp BigQuery - GCP credentials block: GcpCredentials block - Prefect tasks/flows concepts: Tasks and Flows If you share how you plan to run (thread pool vs process pool, local vs. Docker/Kubernetes workers), I can tailor the example to your setup and show the safest injection pattern for it.