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) manifest 是否可用。

---

1. 配置项目（ProjectConfig）

方式	适用场景	必需参数
项目路径	文件可在本地访问	dbt_project_path
仅使用 Manifest	加载 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）

根据约束选择一种加载模式：

加载模式	适用场景	必需输入	限制
dbt_manifest	大型项目；容器化执行；速度最快	ProjectConfig.manifest_path	远程 manifest 需要 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。

选择一种执行模式：

执行模式	适用场景	速度	必需设置
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	Airflow 部署中有 dbt 和适配器，且位于独立的 Python 安装环境中	中等	dbt_executable_path
AIRFLOW_ASYNC	BigQuery + 长时间运行的转换	快	Airflow ≥2.8；provider 依赖
KUBERNETES	将 Airflow 与 dbt 隔离	中等	Airflow ≥2.8；provider 依赖
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 连接 + 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）
• 如果下游 DAG 基于 Cosmos 资源调度，请使用 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 数据集）：

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

Cosmos ≥1.10（Airflow 3 资源）：

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）	一个或多个项目	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