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 互換性が導入されました。

# インストール済みバージョンを確認
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 ランタイム/イメージにインストールしてください。

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 の設定 (LOCAL のみ)

重要: Cosmos 付きの dbt Fusion は ExecutionMode.LOCAL と Fusion バイナリを指す dbt_executable_path が必要です。

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 でパースされること (インポート/パース時エラーがないこと)
• マニュアル実行がターゲットウェアハウスに対して成功すること (少なくとも 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 テスト