cosmos-dbt-fusion

用途

本技能用于在 Astronomer Cosmos 中运行 dbt Fusion 项目。它仅覆盖 Fusion 特有的约束与配置步骤，帮助你在 Airflow 中正确集成 dbt Fusion。

前提确认

开始前需确认两点：

dbt 引擎为 Fusion（而非 Core；若为 Core 应使用 cosmos-dbt-core）。
数据仓库仅支持 Snowflake、Databricks、BigQuery 和 Redshift（预览阶段）。

版本说明：dbt Fusion 支持自 Cosmos 1.11.0 引入，要求 Cosmos ≥1.11。

Fusion 特有约束

不支持异步：AIRFLOW_ASYNC 不受支持。
不支持 virtualenv：Fusion 是二进制文件，不是 Python 包。
仓库支持：预览阶段支持 Snowflake、Databricks、BigQuery 和 Redshift。

实施步骤

1. 确认 Cosmos 版本

Cosmos 1.11.0 引入了 dbt Fusion 兼容性。

检查已安装版本：

pip show astronomer-cosmos

如需安装或升级：

pip install "astronomer-cosmos>=1.11.0"

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

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

dbt Fusion 不随 Cosmos 或 dbt Core 捆绑，需安装到 Airflow 运行时/镜像中。

确定安装位置（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,  # 或 DBT_MANIFEST, DBT_LS
)

4. 配置仓库连接（ProfileConfig）

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）

dbt Fusion 与 Cosmos 配合需要 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",  # 必需：Fusion 二进制路径
    # execution_mode 默认为 LOCAL - 不要更改
)

6. 配置项目（ProjectConfig）

from cosmos import ProjectConfig

_project_config = ProjectConfig(
    dbt_project_path="/path/to/dbt/project",
    # manifest_path="/path/to/manifest.json",  # 用于 dbt_manifest 加载模式
    # install_dbt_deps=False,  # 若依赖已在 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 二进制
)

_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

技能介绍

用途