cosmos-dbt-fusion

Cosmos + dbt Fusion：实施清单

按顺序执行步骤。本技能仅涵盖 Fusion 特有的限制。

版本说明：Cosmos 从 1.11.0 开始支持 dbt Fusion。需要 Cosmos ≥1.11。

参考：有关 ProfileConfig、operator_args 和 Airflow 3 兼容性的详细信息，请参阅 reference/cosmos-config.md。

开始之前，请确认：(1) dbt 引擎 = Fusion（而非 Core → 使用 cosmos-dbt-core）；(2) 数据仓库 = 仅限 Snowflake、Databricks、Bigquery 和 Redshift。

Fusion 特有的限制

限制	详情
不支持异步	不支持 AIRFLOW_ASYNC
不支持虚拟环境	Fusion 是二进制文件，而非 Python 包
数据仓库支持	预览期间支持 Snowflake、Databricks、Bigquery 和 Redshift 支持

---

1. 确认 Cosmos 版本

重要：Cosmos 1.11.0 引入了对 dbt Fusion 的兼容性。

# Check installed version
pip show astronomer-cosmos

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

验证：pip show astronomer-cosmos 报告的版本 ≥ 1.11.0

---

2. 安装 dbt Fusion 二进制文件（必需）

dbt Fusion 不随 Cosmos 或 dbt Core 一起提供。请将其安装到 Airflow 运行时/镜像中。

确定 Fusion 二进制文件的安装位置（Dockerfile / 基础镜像 / 运行时）。

Dockerfile 安装示例

USER root
RUN apt-get update && apt-get install -y curl
ENV SHELL=/bin/bash
RUN curl -fsSL https://public.cdn.getdbt.com/fs/install/install.sh | sh -s -- --update
USER astro

常见安装路径

环境	典型路径
Astro Runtime	/home/astro/.local/bin/dbt
系统范围	/usr/local/bin/dbt

验证：所选路径下存在 dbt 二进制文件，且 dbt --version 执行成功。

---

3. 选择解析策略（RenderConfig）

解析策略与 dbt Core 相同。选择一个：

加载模式	适用场景	必需输入
dbt_manifest	大型项目；解析速度最快	ProjectConfig.manifest_path
dbt_ls	选择器复杂；需要 dbt 原生选择	调度器可访问 Fusion 二进制文件
automatic	简单设置；由 Cosmos 自动选择	（无）

from cosmos import RenderConfig, LoadMode

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

---

4. 配置数据仓库连接（ProfileConfig）

参考：有关完整的 ProfileConfig 选项和示例，请参阅 reference/cosmos-config.md。

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. 配置 ExecutionConfig（仅 LOCAL）

重要：Cosmos 中的 dbt Fusion 必须使用 ExecutionMode.LOCAL，并将 dbt_executable_path 指向 Fusion 二进制文件。

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. 配置项目（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. 组装 DAG / TaskGroup

选项 A：DbtDag（独立）

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",
)

选项 B：DbtTaskGroup（在现有 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. 最终验证

完成前，请确认：

• Cosmos 版本：≥1.11.0
• 已安装 Fusion 二进制文件：路径存在且可执行
• 支持的数据仓库：仅限 Snowflake、Databricks、Bigquery 或 Redshift
• 密钥处理：使用 Airflow 连接或环境变量，而非明文

故障排除

如果用户报告启用 Fusion 后 dbt Core 出现回归：

AIRFLOW__COSMOS__PRE_DBT_FUSION=1

用户必须测试

• DAG 可在 Airflow UI 中解析（没有导入/解析时错误）
• 针对目标数据仓库的手动运行成功（至少运行一个模型）

---

参考

• Cosmos dbt Fusion 文档：https://astronomer.github.io/astronomer-cosmos/configuration/dbt-fusion.html
• dbt Fusion 安装：https://docs.getdbt.com/docs/core/pip-install#dbt-fusion

---

相关技能

• cosmos-dbt-core：适用于 dbt Core 项目（非 Fusion）
• authoring-dags：DAG 编写的一般模式
• testing-dags：创建后测试 DAG