Great question! Here’s a straightforward way to orchestrate a dbt Core project that targets BigQuery with Prefect (assuming Prefect 3.x). High-level approach - Use the prefect-dbt collection to run dbt CLI commands (e.g., build/seed/test). - Point dbt at a BigQuery profile (profiles.yml) that uses service account auth or ADC (Google’s Application Default Credentials). - Ensure your runtime environment (Docker image, Cloud Run, K8s, etc.) includes dbt-bigquery and prefect-dbt. What you’ll need - A dbt project checked into your repo (models, dbt_project.yml). - A dbt BigQuery profile. Two common options: - profiles.yml in your repo that uses env vars for secrets - or an ephemeral profile created at runtime (advanced; optional) - A service account key available to the runtime (set GOOGLE_APPLICATION_CREDENTIALS to a keyfile path) or ADC on the infra. Install dependencies in your runtime - Include these in your image/environment: - prefect - prefect-dbt - dbt-bigquery Example dbt profiles.yml for BigQuery Place this in your repo (e.g., ./profiles/profiles.yml) and reference it from Prefect via the profiles_dir parameter. It uses env vars so you avoid checking keys into source.

my_dbt_project:
  target: prod
  outputs:
    prod:
      type: bigquery
      method: service-account
      project: "{{ env_var('GCP_PROJECT') }}"
      dataset: "analytics"
      keyfile: "{{ env_var('GOOGLE_APPLICATION_CREDENTIALS') }}"
      location: "US"
      threads: 4

Then set: - GCP_PROJECT to your GCP project ID - GOOGLE_APPLICATION_CREDENTIALS to the path of a mounted service account JSON key (or rely on ADC and switch method to oauth) Prefect flow using prefect-dbt This flow will run dbt build (which seeds, runs models, and tests). You can also call run_dbt_seed/run_dbt_test separately if you prefer.

from prefect import flow
from prefect_dbt.cli.commands import run_dbt_build  # or run_dbt_seed, run_dbt_test

@flow(name="dbt-bigquery-build")
def dbt_bigquery_flow(
    project_dir: str = "./dbt_project",
    profiles_dir: str = "./profiles",
    target: str = "prod",
    additional_args: list[str] | None = None,
):
    # Make sure env vars like GCP_PROJECT and GOOGLE_APPLICATION_CREDENTIALS
    # are set in your execution environment before this runs.
    extra_args = ["--target", target]
    if additional_args:
        extra_args.extend(additional_args)

    result = run_dbt_build(
        project_dir=project_dir,
        profiles_dir=profiles_dir,
        extra_command_args=extra_args,
        stream_output=True,  # stream dbt logs to Prefect
    )
    return result

if __name__ == "__main__":
    dbt_bigquery_flow()

Notes - project_dir should point to your dbt project root (where dbt_project.yml lives). - profiles_dir should point to a directory containing your profiles.yml. If you set DBT_PROFILES_DIR in the environment, you can omit profiles_dir. Deploying and running - Programmatic deployment (recommended in Prefect 3.x): load your flow from source and call deploy. For example: - Have your worker pick up code from a Git repo and ensure the worker image includes prefect-dbt and dbt-bigquery. - Use flow.from_source(...).deploy(...) in a small bootstrap script or do it via the UI. - Scheduling: add a schedule via the UI or programmatically when deploying (e.g., a daily cron). If you want help wiring a code-based schedule, I can share a snippet based on your exact setup. Auth options for BigQuery - Service account keyfile - Mount the JSON key in your runtime and set GOOGLE_APPLICATION_CREDENTIALS to its path. - profiles.yml uses method: service-account and keyfile: "{{ env_var('GOOGLE_APPLICATION_CREDENTIALS') }}" - Application Default Credentials (recommended in GCP environments like Cloud Run/GKE) - Use method: oauth in your profiles.yml. - Ensure your runtime’s service account has BigQuery permissions. Alternative: dbt Cloud

If you use dbt Cloud, prefect-dbt also supports triggering dbt Cloud jobs (credentials and job configs differ). Happy to share that pattern if that’s your stack. Would you like me to tailor this to your environment? - Are you using dbt Core or dbt Cloud? - Where will your flow run (local, Docker, K8s, Cloud Run, etc.)? - How do you plan to provide BigQuery credentials (service account keyfile vs ADC)? - Where does your dbt project live (Git repo path)?

Short answer: Yes—with dbt Core you can avoid a physical keyfile, but it’s dbt’s templating (env_var in profiles.yml) that does the heavy lifting. In practice, there are two reliable patterns: Option A: Use keyfile_json in your dbt BigQuery profile (if your dbt-bigquery version supports it) - In profiles.yml, reference the service account JSON via an env var:

my_dbt_project:
  target: prod
  outputs:
    prod:
      type: bigquery
      method: service-account
      keyfile_json: "{{ env_var('GCP_SA_JSON') }}"
      project: "{{ env_var('GCP_PROJECT') }}"
      dataset: "analytics"
      location: "US"
      threads: 4

- Set GCP_SA_JSON to the raw JSON string (single-line). Many teams base64-encode it to avoid quoting/newline issues and then decode before setting. - Then in Prefect, just ensure those env vars are available to the flow run (via your deployment/work pool/worker environment). Option B: Write the JSON to a temp file at runtime, then point dbt to it - This works across all dbt-bigquery versions and avoids relying on keyfile_json support. - Store the JSON in a Prefect/worker env var or secret, write it to /tmp, and set GOOGLE_APPLICATION_CREDENTIALS (or use keyfile in profiles.yml). Example with prefect-dbt and a temp keyfile

import os, tempfile
from prefect import flow
from prefect_dbt.cli.commands import run_dbt_build

@flow
def dbt_bigquery_flow(
    project_dir: str = "./dbt_project",
    profiles_dir: str = "./profiles",
    target: str = "prod",
):
    # Expect the raw JSON in an env var (e.g., injected via your deployment or worker)
    sa_json = os.environ["GCP_SA_JSON"]
    project_id = os.environ["GCP_PROJECT"]

    # Write to a temp file so dbt can use it
    with tempfile.NamedTemporaryFile("w", suffix=".json", delete=False) as f:
        f.write(sa_json)
        key_path = f.name

    # Make credentials visible to dbt
    os.environ["GOOGLE_APPLICATION_CREDENTIALS"] = key_path
    os.environ["GCP_PROJECT"] = project_id

    # Run dbt
    run_dbt_build(
        project_dir=project_dir,
        profiles_dir=profiles_dir,     # profiles.yml can use method: service-account + keyfile: "{{ env_var('GOOGLE_APPLICATION_CREDENTIALS') }}"
        extra_command_args=["--target", target],
        stream_output=True,
    )

Notes - If you prefer an entirely ephemeral profile (no profiles.yml), prefect-dbt’s DbtCliProfile + BigQueryTargetConfigs can build the profile at runtime. You’d still either pass a keyfile path or, if supported by your dbt-bigquery version, the JSON content. - If you’re running on GCP-managed infra (Cloud Run, GKE), consider using Application Default Credentials (method: oauth) instead of a service account key—no JSON needed. If you tell me your runtime (Docker/K8s/Cloud Run/local) and whether you want to use env vars or Prefect variables/secrets, I can share a tailored snippet for that setup.