Overview

Converts Prefect dbt tasks into Orchestra pipeline YAML. Two dbt variants exist:

• dbt Core (DbtCoreOperation, ShellOperation running dbt CLI): integration: DBT_CORE, integration_job: DBT_CORE_EXECUTE
• dbt Cloud (DbtCloudJob): integration: DBT, integration_job: DBT_RUNJOB

The commands list from Prefect is joined into a single semicolon-delimited string. Warehouse credentials and Git repo details go on the Orchestra dbt Core connection, not in YAML.

Parameter Mapping

Orchestra YAML Structure

dbt Core:

version: v1
name: dbt-flow
pipeline:
  stage-001:
    tasks:
      task-001:
        integration: DBT_CORE
        integration_job: DBT_CORE_EXECUTE
        name: dbt_daily_build
        connection: data_dbt_bigquery_prod_12345
        parameters:
          commands: 'dbt seed; dbt build --select tag:daily --target prod;'
          package_manager: PIP
          python_version: '3.12'
          project_dir: null   # optional — subdirectory holding dbt_project.yml, e.g. 'dbt' in a monorepo
        depends_on: []
        condition: null
        tags: []

dbt Cloud:

version: v1
name: dbt-cloud-flow
pipeline:
  stage-001:
    tasks:
      task-001:
        integration: DBT
        integration_job: DBT_RUNJOB
        name: dbt_cloud_job
        connection: dbt_cloud_prod_12345
        parameters:
          job_id: '123456'
        depends_on: []
        condition: null
        tags: []

Conversion Steps

• Determine dbt variant: DbtCoreOperation / ShellOperation → DBT_CORE; DbtCloudJob → DBT
• dbt Core: join commands list with '; ' and append trailing ; into parameters.commands
• dbt Core: choose parameters.package_manager (PIP/POETRY/UV) to match project setup — if it can't be inferred, add a # MANUAL: comment flagging it for the user to confirm/select when they set up the dbt Core connection, rather than silently defaulting to PIP
• dbt Core: choose parameters.python_version ('3.11', '3.12', '3.13' or '3.14') — match the source project's Python version where it's supported, rather than defaulting to '3.12' regardless
• dbt Core: carry project_dir over to parameters.project_dir (a real task parameter — don't drop it); move profiles.yml and warehouse credentials to the Orchestra dbt Core connection
• dbt Cloud: set parameters.job_id to the numeric job ID string; set account_id on the connection
• Set connection: to the appropriate Orchestra connection name
• Wire depends_on: if this task follows a Fivetran sync or other upstream task
• Add an alerts: block if the Prefect flow had on_failure/on_completion hooks

Before / After Example

Prefect (before)

from prefect import flow
from prefect_dbt.cli import DbtCoreOperation

@flow
def dbt_flow():
    with DbtCoreOperation(
        commands=["dbt seed", "dbt build --select tag:daily --target prod"],
        project_dir="dbt_project",
    ) as dbt_op:
        dbt_op.run()

Orchestra YAML (after)

version: v1
name: dbt-flow
pipeline:
  stage-001:
    tasks:
      task-001:
        integration: DBT_CORE
        integration_job: DBT_CORE_EXECUTE
        name: dbt_daily_build
        connection: data_dbt_bigquery_prod_12345
        parameters:
          commands: 'dbt seed; dbt build --select tag:daily --target prod;'
          package_manager: PIP
          python_version: '3.12'
          project_dir: dbt_project   # from DbtCoreOperation(project_dir="dbt_project") in the source above
        depends_on: []
        condition: null
        tags: []

Warehouse Credentials: Link, Don't Copy

A dbt Core connection needs warehouse credentials, but they don't have to be re-typed into it. If the warehouse already exists as its own Orchestra connection — the Snowflake/Postgres/BigQuery connection the rest of the pipeline already uses — link it under Linked connections on the dbt Core connection's form rather than pasting the same password into that connection's Secrets. One credential, one place to rotate it.

Linked fields arrive as environment variables named <INTEGRATION>__<CONNECTION_ID>__<FIELD>: every part upper-cased, each run of non-alphanumeric characters collapsed to a single _, and the parts joined by a double underscore. <CONNECTION_ID> is the id the YAML's connection: field uses (it survives renames), and <FIELD> is the field's ID from that integration's Authentication parameters table. Read them in profiles.yml with dbt's env_var():

my_project:
  target: prod
  outputs:
    prod:
      type: snowflake
      account: "{{ env_var('SNOWFLAKE__PROD_12345__SNOWFLAKE_ACCOUNT') }}"
      user: "{{ env_var('SNOWFLAKE__PROD_12345__SNOWFLAKE_USER') }}"
      password: "{{ env_var('SNOWFLAKE__PROD_12345__SNOWFLAKE_PASSWORD') }}"

• A Snowflake connection using key-pair auth exports SNOWFLAKE__PROD_12345__SNOWFLAKE_PRIVATE_KEY instead of a password — check the integration's field IDs rather than assuming password.
• A linked connection that has been deleted, or whose credentials can't be decrypted, is skipped silently — its variables are simply absent, so don't write a profile that assumes one is set.
• Linking is also what opens an approved private connection for a task on Orchestra-hosted compute; private connections are not opened for connections that aren't linked.

Full reference: Linked connections.

Gotchas

• DbtCoreOperation.commands is a Python list — join with semicolons for Orchestra ('dbt seed; dbt build;'); a missing trailing ; can cause parse errors
• project_dir is a real task parameter — carry DbtCoreOperation(project_dir=...)'s value over to parameters.project_dir instead of dropping it. Only null it if the dbt project genuinely sits at the repo root.
• profiles.yml and warehouse credentials go on the Orchestra dbt Core connection, never in YAML; if the warehouse is already its own Orchestra connection, link it rather than copying its secrets across (see above)
• DbtCloudJob (Cloud) uses a different integration pair: integration: DBT, integration_job: DBT_RUNJOB, parameters: {job_id: "123456"}
• ShellOperation(commands=["dbt build"]) is treated identically to DbtCoreOperation
• --select, --exclude, and --target flags belong inside the commands string, not as separate parameters
• Can't infer package_manager: don't just guess PIP and move on. Flag it instead:

  package_manager: PIP  # MANUAL: could not detect pip/poetry/uv from the source — confirm/select the correct package manager when setting up the dbt Core connection

  This surfaces as a Manual Review item in the migration checklist, so the user resolves it alongside connection setup rather than deploying a silently-wrong environment.

References

• https://docs.getorchestra.io/docs/integrations/dbt_core
• https://docs.getorchestra.io/docs/integrations/dbt_core#linked-connections
• https://docs.getorchestra.io/docs/integrations/dbt
• https://prefecthq.github.io/prefect-dbt/

Adding Alerts

See prefect-alerts-to-orchestra for all notification patterns.