OverviewHistoryStatsSecurity
npx skills add ...
Documentation
SKILL.md
npx skills add astronomer/agents --skill cosmos-dbt-fusion
Run a dbt Fusion project with Astronomer Cosmos. Use when running a dbt Fusion project with Astronomer Cosmos (Cosmos 1.11+, ExecutionMode.LOCAL on Snowflake/Databricks). Before implementing, verify dbt engine is Fusion (not Core), the warehouse is supported, and local execution is acceptable. Does not cover dbt Core.
npx skills add astronomer/agents --skill cosmos-dbt-fusion
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.
| 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 |
CRITICAL: Cosmos 1.11.0 introduced dbt Fusion compatibility.
Validate: pip show astronomer-cosmos reports version ≥ 1.11.0
dbt Fusion is NOT bundled with Cosmos or dbt Core. Install it into the Airflow runtime/image.
Determine where to install the Fusion binary (Dockerfile / base image / runtime).
| Environment | Typical path |
|---|---|
| Astro Runtime | /home/astro/.local/bin/dbt |
| System-wide | /usr/local/bin/dbt |
Validate: The dbt binary exists at the chosen path and dbt --version succeeds.
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) |
Reference: See reference/cosmos-config.md for full ProfileConfig options and examples.
CRITICAL: dbt Fusion with Cosmos requires
ExecutionMode.LOCALwithdbt_executable_pathpointing to the Fusion binary.
Before finalizing, verify:
If user reports dbt Core regressions after enabling Fusion:
from cosmos import RenderConfig, LoadMode
_render_config = RenderConfig(
load_method=LoadMode.AUTOMATIC, # or DBT_MANIFEST, DBT_LS
)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",
),
)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
)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
)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",
)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()AIRFLOW__COSMOS__PRE_DBT_FUSION=1