cosmos-dbt-core

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

手順に沿って順番に実行してください。ユーザーの制約を満たす最もシンプルな設定を優先してください。

バージョンに関する注意：このスキルは Cosmos 1.11+ および Airflow 3.x を対象としています。ユーザーが Airflow 2.x を使用している場合は、それに応じてインポートを調整してください（付録 A を参照）。

リファレンス：最新の安定版：https://pypi.org/project/astronomer-cosmos/

開始前に、次の点を確認してください：（1）dbt エンジン = Core（Fusion の場合は cosmos-dbt-fusion を使用）、（2）ウェアハウスの種類、（3）Airflow のバージョン、（4）実行環境（Airflow 環境／venv／コンテナ）、（5）DbtDag、DbtTaskGroup、個別オペレーターのどれを使うか、（6）マニフェストが利用可能かどうか。

---

1. プロジェクトの設定（ProjectConfig）

方法	使用する状況	必須パラメーター
プロジェクトパス	ファイルをローカルで利用できる	dbt_project_path
マニフェストのみ	dbt_manifest の読み込み	manifest_path + project_name

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
    # project_name="my_project",  # if using manifest_path without dbt_project_path
    # install_dbt_deps=False,  # if deps precomputed in CI
)

2. 解析方法の選択（RenderConfig）

制約に基づいて、読み込みモードを1つ選択してください。

読み込みモード	使用する状況	必須の入力	制約
dbt_manifest	大規模なプロジェクト、コンテナ化された実行、最速	ProjectConfig.manifest_path	リモートマニフェストには manifest_conn_id が必要です
dbt_ls	複雑なセレクター、dbt ネイティブの選択が必要	dbt がインストールされている、または dbt_executable_path	コンテナ化された実行でも使用できます
dbt_ls_file	解析のたびに dbt_ls を実行せずに dbt_ls による選択を行う	RenderConfig.dbt_ls_path	select／exclude は機能しません
automatic（デフォルト）	シンプルな構成、Cosmos に選択を任せる	（なし）	フォールバック：manifest → dbt_ls → custom

重要：コンテナ化された実行（DOCKER／KUBERNETES など）

from cosmos import RenderConfig, LoadMode

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

---

3. 実行モードの選択（ExecutionConfig）

リファレンス：モードごとの詳細な設定例については、reference/cosmos-config.md を参照してください。

実行モードを1つ選択してください。

実行モード	使用する状況	速度	必要な設定
WATCHER	最速、単一の dbt build の可視性	最速	環境内に dbt アダプターがある、または dbt_executable_path を指定する、または dbt Fusion を使用する
WATCHER_KUBERNETES	最速の分離方式、単一の dbt build の可視性	高速	コンテナに dbt がインストールされている
LOCAL + DBT_RUNNER	dbt とアダプターが Airflow と同じ Python インストール環境にある	高速	requirements.txt に dbt 1.5+ が必要
LOCAL + SUBPROCESS	dbt とアダプターが Airflow デプロイメント内の独立した Python インストール環境で利用可能	中速	dbt_executable_path
AIRFLOW_ASYNC	BigQuery + 長時間実行される変換	高速	Airflow ≥2.8、プロバイダーの依存関係
KUBERNETES	Airflow と dbt を分離する	中速	Airflow ≥2.8、プロバイダーの依存関係
VIRTUALENV	イメージを変更できない、実行時の venv	低速	operator_args 内の py_requirements
その他のコンテナ化方式	Airflow と dbt の分離をサポートする	中速	コンテナ設定

from cosmos import ExecutionConfig, ExecutionMode

_execution_config = ExecutionConfig(
    execution_mode=ExecutionMode.WATCHER,  # or LOCAL, VIRTUALENV, AIRFLOW_ASYNC, KUBERNETES, etc.
)

---

4. ウェアハウス接続の設定（ProfileConfig）

リファレンス：詳細な ProfileConfig オプションとすべての ProfileMapping クラスについては、reference/cosmos-config.md を参照してください。

オプション A：Airflow Connection + ProfileMapping（推奨）

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",
        profile_args={"schema": "my_schema"},
    ),
)

オプション B：既存の profiles.yml

重要：シークレットをハードコードしないでください。環境変数を使用してください。

from cosmos import ProfileConfig

_profile_config = ProfileConfig(
    profile_name="my_profile",
    target_name="dev",
    profiles_yml_filepath="/path/to/profiles.yml",
)

---

5. テスト動作の設定（RenderConfig）

リファレンス：テストオプションの詳細については、reference/cosmos-config.md を参照してください。

TestBehavior	動作
AFTER_EACH（デフォルト）	各モデルの直後にテストを実行します（デフォルト）
BUILD	run と test を単一の dbt build にまとめます
AFTER_ALL	すべてのモデルの完了後に、すべてのテストを実行します
NONE	テストをスキップします

from cosmos import RenderConfig, TestBehavior

_render_config = RenderConfig(
    test_behavior=TestBehavior.AFTER_EACH,
)

---

6. operator_args の設定

参考：operator_args のオプションの詳細については、reference/cosmos-config.md を参照してください。

_operator_args = {
    # BaseOperator params
    "retries": 3,

    # Cosmos-specific params
    "install_deps": False,
    "full_refresh": False,
    "quiet": True,

    # Runtime dbt vars (XCom / params)
    "vars": '{"my_var": "{{ ti.xcom_pull(task_ids=\'pre_dbt\') }}"}',
}

---

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()
_render_config = RenderConfig()

my_cosmos_dag = DbtDag(
    dag_id="my_cosmos_dag",
    project_config=_project_config,
    profile_config=_profile_config,
    execution_config=_execution_config,
    render_config=_render_config,
    operator_args={},
    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, RenderConfig
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()
_render_config = RenderConfig()

@dag(start_date=datetime(2025, 1, 1), schedule="@daily")
def my_dag():
    @task
    def pre_dbt():
        return "some_value"

    dbt = DbtTaskGroup(
        group_id="dbt_project",
        project_config=_project_config,
        profile_config=_profile_config,
        execution_config=_execution_config,
        render_config=_render_config,
    )

    @task
    def post_dbt():
        pass

    chain(pre_dbt(), dbt, post_dbt())

my_dag()

オプション C：Cosmos のオペレーターを直接使用する

import os
from datetime import datetime
from pathlib import Path
from typing import Any

from airflow import DAG

try:
    from airflow.providers.standard.operators.python import PythonOperator
except ImportError:
    from airflow.operators.python import PythonOperator

from cosmos import DbtCloneLocalOperator, DbtRunLocalOperator, DbtSeedLocalOperator, ProfileConfig
from cosmos.io import upload_to_aws_s3

DEFAULT_DBT_ROOT_PATH = Path(__file__).parent / "dbt"
DBT_ROOT_PATH = Path(os.getenv("DBT_ROOT_PATH", DEFAULT_DBT_ROOT_PATH))
DBT_PROJ_DIR = DBT_ROOT_PATH / "jaffle_shop"
DBT_PROFILE_PATH = DBT_PROJ_DIR / "profiles.yml"
DBT_ARTIFACT = DBT_PROJ_DIR / "target"

profile_config = ProfileConfig(
    profile_name="default",
    target_name="dev",
    profiles_yml_filepath=DBT_PROFILE_PATH,
)


def check_s3_file(bucket_name: str, file_key: str, aws_conn_id: str = "aws_default", **context: Any) -> bool:
    """Check if a file exists in the given S3 bucket."""
    from airflow.providers.amazon.aws.hooks.s3 import S3Hook

    s3_key = f"{context['dag'].dag_id}/{context['run_id']}/seed/0/{file_key}"
    print(f"Checking if file {s3_key} exists in S3 bucket...")
    hook = S3Hook(aws_conn_id=aws_conn_id)
    return hook.check_for_key(key=s3_key, bucket_name=bucket_name)


with DAG("example_operators", start_date=datetime(2024, 1, 1), catchup=False) as dag:
    seed_operator = DbtSeedLocalOperator(
        profile_config=profile_config,
        project_dir=DBT_PROJ_DIR,
        task_id="seed",
        dbt_cmd_flags=["--select", "raw_customers"],
        install_deps=True,
        append_env=True,
    )

    check_file_uploaded_task = PythonOperator(
        task_id="check_file_uploaded_task",
        python_callable=check_s3_file,
        op_kwargs={
            "aws_conn_id": "aws_s3_conn",
            "bucket_name": "cosmos-artifacts-upload",
            "file_key": "target/run_results.json",
        },
    )

    run_operator = DbtRunLocalOperator(
        profile_config=profile_config,
        project_dir=DBT_PROJ_DIR,
        task_id="run",
        dbt_cmd_flags=["--models", "stg_customers"],
        install_deps=True,
        append_env=True,
    )

    clone_operator = DbtCloneLocalOperator(
        profile_config=profile_config,
        project_dir=DBT_PROJ_DIR,
        task_id="clone",
        dbt_cmd_flags=["--models", "stg_customers", "--state", DBT_ARTIFACT],
        install_deps=True,
        append_env=True,
    )

    seed_operator >> run_operator >> clone_operator
    seed_operator >> check_file_uploaded_task

個々の Cosmos タスクの依存関係の設定

from cosmos import DbtDag, DbtResourceType
from airflow.sdk import task, chain

with DbtDag(...) as dag:
    @task
    def upstream_task():
        pass

    _upstream = upstream_task()

    for unique_id, dbt_node in dag.dbt_graph.filtered_nodes.items():
        if dbt_node.resource_type == DbtResourceType.SEED:
            my_dbt_task = dag.tasks_map[unique_id]
            chain(_upstream, my_dbt_task)

---

8. 安全性の確認

仕上げる前に、以下を確認してください。

• 実行モードが制約と一致している（AIRFLOW_ASYNC → BigQuery のみ）
• 選択した実行モード用のウェアハウスアダプターがインストールされている
• シークレットは Airflow 接続または環境変数を使用しており、平文ではない
• ロードモードが実行方法と一致している（複雑なセレクター → dbt_ls）
• Cosmos のアセットを基に下流の DAG がスケジュールされる場合、Airflow 3 のアセット URI を使用している（付録 A を参照）

---

付録 A：Airflow 3 との互換性

インポートの違い

Airflow 3.x	Airflow 2.x
from airflow.sdk import dag, task	from airflow.decorators import dag, task
from airflow.sdk import chain	from airflow.models.baseoperator import chain

アセット／データセットの URI 形式の変更

Cosmos ≤1.9（Airflow 2 の Datasets）：

postgres://0.0.0.0:5434/postgres.public.orders

Cosmos ≥1.10（Airflow 3 の Assets）：

postgres://0.0.0.0:5434/postgres/public/orders

重要：Airflow 3 にアップグレードする際は、アセット URI を更新してください。

---

付録 B：運用に関する追加設定

キャッシュ

Cosmos は解析を高速化するためにアーティファクトをキャッシュします。デフォルトで有効です。

参考：https://astronomer.github.io/astronomer-cosmos/configuration/caching.html

メモリ最適化インポート

AIRFLOW__COSMOS__ENABLE_MEMORY_OPTIMISED_IMPORTS=True

有効にすると：

from cosmos.airflow.dag import DbtDag  # instead of: from cosmos import DbtDag

オブジェクトストレージへのアーティファクトのアップロード

AIRFLOW__COSMOS__REMOTE_TARGET_PATH=s3://bucket/target_dir/
AIRFLOW__COSMOS__REMOTE_TARGET_PATH_CONN_ID=aws_default

from cosmos.io import upload_to_cloud_storage

my_dag = DbtDag(
    # ...
    operator_args={"callback": upload_to_cloud_storage},
)

dbt ドキュメントのホスティング

Cosmos は Airflow の UI で dbt ドキュメントを提供します。設定は Airflow のメジャーバージョンによって異なります（UI プラグインシステムが異なるためです）。単一プロジェクトと複数プロジェクトを自由に選べるわけではありません。

Airflow	設定	適用範囲	対応バージョン
2（FAB プラグイン）	DBT_DOCS_DIR（および DBT_DOCS_CONN_ID、DBT_DOCS_INDEX_FILE_NAME）	単一プロジェクト	Cosmos 1.4.0 以降
3.1 以降（FastAPI）	DBT_DOCS_PROJECTS（JSON）	1 つ以上のプロジェクト	Cosmos 1.11.0 以降

Airflow 2：

AIRFLOW__COSMOS__DBT_DOCS_DIR="path/to/docs"                   # local path or S3/GCS/Azure/HTTP URI; defaults to the dbt target/ folder
AIRFLOW__COSMOS__DBT_DOCS_CONN_ID="my_conn_id"                 # optional; for cloud storage
AIRFLOW__COSMOS__DBT_DOCS_INDEX_FILE_NAME="static_index.html"  # optional; only if docs built with --static

Airflow 3.1 以降：

AIRFLOW__COSMOS__DBT_DOCS_PROJECTS='{
    "my_project": {
        "dir": "s3://bucket/docs/",
        "index": "index.html",
        "conn_id": "aws_default",
        "name": "My Project"
    }
}'

プロジェクト数ではなく、Airflow のバージョンに基づいて選択してください。単一プロジェクトの設定は Airflow 2 向けです。Cosmos はこれらの設定について非推奨の告知をしていないため、「レガシー」または「非推奨」と表現しないでください。

参考：https://astronomer.github.io/astronomer-cosmos/configuration/hosting-docs.html

---

関連スキル

• cosmos-dbt-fusion：dbt Fusion プロジェクト向け（dbt Core ではありません）
• authoring-dags：DAG 作成の一般的なパターン
• testing-dags：作成後の DAG のテスト