cosmos-dbt-fusion

by astronomer

Configure Astronomer Cosmos for dbt Fusion projects on Snowflake, Databricks, BigQuery, or Redshift with local execution. Requires Cosmos 1.11.0+, dbt Fusion binary installed separately in the Airflow runtime, and ExecutionMode.LOCAL with subprocess invocation Supports three parsing strategies: dbt_manifest (fastest for large projects), dbt_ls (for complex selectors), or automatic (simple setups) Covers ProfileConfig setup for warehouse connections, ProjectConfig for dbt project paths, and... npx skills add https://github.com/astronomer/agents --skill cosmos-dbt-fusion Download ZIPGitHub397

Cosmos + dbt Fusion: Implementation Checklist

Execute steps in order. This skill covers Fusion-specific constraints only.

Version note: dbt Fusion support was introduced in Cosmos 1.11.0. Requires Cosmos ≥1.11.

Reference: See reference/cosmos-config.md for ProfileConfig, operator_args, and Airflow 3 compatibility details.

Before starting, confirm: (1) dbt engine = Fusion (not Core → use cosmos-dbt-core), (2) warehouse = Snowflake, Databricks, Bigquery and Redshift only.

Fusion-Specific Constraints

Constraint Details No async AIRFLOW_ASYNC not supported No virtualenv Fusion is a binary, not a Python package Warehouse support Snowflake, Databricks, Bigquery and Redshift support while in preview

1. Confirm Cosmos Version

CRITICAL: Cosmos 1.11.0 introduced dbt Fusion compatibility.

# Check installed version
pip show astronomer-cosmos

# Install/upgrade if needed
pip install "astronomer-cosmos>=1.11.0"

Validate: pip show astronomer-cosmos reports version ≥ 1.11.0

3. Choose Parsing Strategy (RenderConfig)

Parsing strategy is the same as dbt Core. Pick ONE:

Load mode When to use Required inputs dbt_manifest Large projects; fastest parsing ProjectConfig.manifest_path dbt_ls Complex selectors; need dbt-native selection Fusion binary accessible to scheduler automatic Simple setups; let Cosmos pick (none)

from cosmos import RenderConfig, LoadMode

_render_config = RenderConfig(
 load_method=LoadMode.AUTOMATIC, # or DBT_MANIFEST, DBT_LS
)

4. Configure Warehouse Connection (ProfileConfig)

Reference: See reference/cosmos-config.md for full ProfileConfig options and examples.

from cosmos import ProfileConfig
from cosmos.profiles import SnowflakeUserPasswordProfileMapping

_profile_config = ProfileConfig(
 profile_name="default",
 target_name="dev",
 profile_mapping=SnowflakeUserPasswordProfileMapping(
 conn_id="snowflake_default",
 ),
)

5. Configure ExecutionConfig (LOCAL Only)

CRITICAL: dbt Fusion with Cosmos requires ExecutionMode.LOCAL with dbt_executable_path pointing to the Fusion binary.

from cosmos import ExecutionConfig
from cosmos.constants import InvocationMode

_execution_config = ExecutionConfig(
 invocation_mode=InvocationMode.SUBPROCESS,
 dbt_executable_path="/home/astro/.local/bin/dbt", # REQUIRED: path to Fusion binary
 # execution_mode is LOCAL by default - do not change
)

6. Configure Project (ProjectConfig)

from cosmos import ProjectConfig

_project_config = ProjectConfig(
 dbt_project_path="/path/to/dbt/project",
 # manifest_path="/path/to/manifest.json", # for dbt_manifest load mode
 # install_dbt_deps=False, # if deps precomputed in CI
)

7. Assemble DAG / TaskGroup

Option A: DbtDag (Standalone)

from cosmos import DbtDag, ProjectConfig, ProfileConfig, ExecutionConfig, RenderConfig
from cosmos.profiles import SnowflakeUserPasswordProfileMapping
from pendulum import datetime

_project_config = ProjectConfig(
 dbt_project_path="/usr/local/airflow/dbt/my_project",
)

_profile_config = ProfileConfig(
 profile_name="default",
 target_name="dev",
 profile_mapping=SnowflakeUserPasswordProfileMapping(
 conn_id="snowflake_default",
 ),
)

_execution_config = ExecutionConfig(
 dbt_executable_path="/home/astro/.local/bin/dbt", # Fusion binary
)

_render_config = RenderConfig()

my_fusion_dag = DbtDag(
 dag_id="my_fusion_cosmos_dag",
 project_config=_project_config,
 profile_config=_profile_config,
 execution_config=_execution_config,
 render_config=_render_config,
 start_date=datetime(2025, 1, 1),
 schedule="@daily",
)

Option B: DbtTaskGroup (Inside Existing DAG)

from airflow.sdk import dag, task # Airflow 3.x
# from airflow.decorators import dag, task # Airflow 2.x
from airflow.models.baseoperator import chain
from cosmos import DbtTaskGroup, ProjectConfig, ProfileConfig, ExecutionConfig
from pendulum import datetime

_project_config = ProjectConfig(dbt_project_path="/usr/local/airflow/dbt/my_project")
_profile_config = ProfileConfig(profile_name="default", target_name="dev")
_execution_config = ExecutionConfig(dbt_executable_path="/home/astro/.local/bin/dbt")

@dag(start_date=datetime(2025, 1, 1), schedule="@daily")
def my_dag():
 @task
 def pre_dbt():
 return "some_value"

 dbt = DbtTaskGroup(
 group_id="dbt_fusion_project",
 project_config=_project_config,
 profile_config=_profile_config,
 execution_config=_execution_config,
 )

 @task
 def post_dbt():
 pass

 chain(pre_dbt(), dbt, post_dbt())

my_dag()

8. Final Validation

Before finalizing, verify:

• Cosmos version: ≥1.11.0

• Fusion binary installed: Path exists and is executable

• Warehouse supported: Snowflake, Databricks, Bigquery or Redshift only

• Secrets handling: Airflow connections or env vars, NOT plaintext

Troubleshooting

If user reports dbt Core regressions after enabling Fusion:

AIRFLOW__COSMOS__PRE_DBT_FUSION=1

User Must Test

• The DAG parses in the Airflow UI (no import/parse-time errors)

• A manual run succeeds against the target warehouse (at least one model)

Reference

• Cosmos dbt Fusion docs: https://astronomer.github.io/astronomer-cosmos/configuration/dbt-fusion.html

• dbt Fusion install: https://docs.getdbt.com/docs/core/pip-install#dbt-fusion

Related Skills

• cosmos-dbt-core: For dbt Core projects (not Fusion)

• authoring-dags: General DAG authoring patterns

• testing-dags: Testing DAGs after creation