cosmos-dbt-fusion

Cosmos + dbt Fusion：実装チェックリスト

手順を順番に実行します。このスキルでは、Fusion 固有の制約のみを扱います。

バージョンに関する注意：dbt Fusion のサポートは Cosmos 1.11.0 で導入されました。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 はサポートされていません
virtualenv は非対応	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 と同じです。1 つ選択してください。

読み込みモード	使用する場面	必要な入力
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 を設定する（ローカルのみ）

重要：Cosmos で dbt Fusion を使用するには、dbt_executable_path が Fusion バイナリを指す ExecutionMode.LOCAL が必要です。

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

ユーザーによるテスト

• Airflow UI で DAG が解析される（インポート時／解析時エラーがない）
• ターゲットのウェアハウスに対する手動実行が成功する（少なくとも 1 つのモデル）

---

参考資料

• 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 のテスト