SuperAPI AI 工具集

cosmos-dbt-core

数据与分析

查看项目

更新于 10-14 21:23

技能介绍

本技能是一份 Cosmos + dbt Core 实现清单,按顺序执行各步骤,并优先选择满足约束的最简配置。版本说明:技能针对 Cosmos 1.11+ 与 Airflow 3.x;若用户使用 Airflow 2.x,需要调整导入方式(见附录 A)。参考资料为 PyPI 上的 astronomer-cosmos。

【开始前确认】需要确认:(1) dbt 引擎为 Core(不是 Fusion,Fusion 应使用 cosmos-dbt-fusion);(2) 仓库类型;(3) Airflow 版本;(4) 执行环境(Airflow 环境 / venv / 容器);(5) 选择 DbtDag、DbtTaskGroup 还是个别的算子;(6) manifest 是否可用。

【一、配置项目 ProjectConfig】可按项目路径(本地文件可用时,参数 dbt_project_path)或仅用 manifest(dbt_manifest 加载方式,参数 manifest_path + project_name)。示例从 cosmos 导入 ProjectConfig 并设置 dbt_project_path,可附带 manifest_path、project_name、install_dbt_deps(CI 中已预计算依赖时)。

【二、选择解析策略 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 等)时需重点注意。

【三、选择执行模式 ExecutionConfig】从以下模式中选一种:WATCHER(最快、单次 dbt build 可见性;需要环境中含 dbt adapter 或 dbt_executable_path 或 dbt Fusion);WATCHER_KUBERNETES(最快隔离方式、单次 dbt build 可见性;容器内安装 dbt);LOCAL + DBT_RUNNER(dbt 与 adapter 与 Airflow 同一 Python 安装;需要 requirements.txt 中 dbt 1.5+);LOCAL + SUBPROCESS(dbt 与 adapter 在 Airflow 部署中的隔离 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 隔离;需要容器配置)。详细配置示例见 reference/cosmos-config.md。

【四、配置仓库连接 ProfileConfig】推荐方式 A:Airflow 连接 + ProfileMapping,例如使用 SnowflakeUserPasswordProfileMapping,设置 profile_name、target_name、conn_id 与 profile_args。方式 B:使用已有 profiles.yml,设置 profiles_yml_filepath,注意不要硬编码密钥,应使用环境变量。所有 ProfileMapping 类与详细选项见 reference/cosmos-config.md。

【五、配置测试行为 RenderConfig】TestBehavior 可选:AFTER_EACH(默认,每个模型后立即测试)、BUILD(将 run + test 合并为单次 dbt build)、AFTER_ALL(所有模型完成后统一测试)、NONE(跳过测试)。

【六、配置 operator_args】可设置基础算子参数如 retries,以及 Cosmos 特有参数 install_deps、full_refresh、quiet,以及运行时 dbt vars(支持 XCom / params 模板)。

【七、组装 DAG / TaskGroup】提供三种方式。方式 A:DbtDag 独立 DAG,传入 dag_id、project_config、profile_config、execution_config、render_config、operator_args、start_date、schedule。方式 B:在已有 DAG 内使用 DbtTaskGroup,注意 Airflow 3.x 从 airflow.sdk 导入 dag/task,Airflow 2.x 从 airflow.decorators 导入,并使用 chain 串联前置任务、dbt 任务组与后置任务。方式 C:直接使用 Cosmos 算子(如 DbtSeedLocalOperator、DbtRunLocalOperator、DbtCloneLocalOperator),配合 ProfileConfig 与 profiles.yml 路径,可设置 dbt_cmd_flags、install_deps、append_env 等,并通过 >> 设置依赖。还介绍了如何对个别 Cosmos 任务设置依赖:遍历 dag.dbt_graph.filtered_nodes,按 resource_type(如 SEED)通过 dag.tasks_map 取任务并 chain。

【八、安全检查】最终确认前需验证:执行模式与约束匹配(AIRFLOW_ASYNC 仅支持 BigQuery);所选执行模式对应的仓库 adapter 已安装;密钥通过 Airflow 连接或环境变量提供而非明文;加载模式与执行匹配(复杂选择器使用 dbt_ls);如下游 DAG 基于 Cosmos asset 调度,需使用 Airflow 3 asset URI。

【附录 A:Airflow 3 兼容性】导入差异:Airflow 3.x 使用 from airflow.sdk import dag, task 与 from airflow.sdk import chain;Airflow 2.x 使用 from airflow.decorators import dag, task 与 from airflow.models.baseoperator import chain。Asset/Dataset 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 时需更新 asset URI。

【附录 B:运维附加项】缓存:Cosmos 默认启用 artifacts 缓存以加速解析。内存优化导入:设置 AIRFLOW__COSMOS__ENABLE_MEMORY_OPTIMISED_IMPORTS=True 后,从 cosmos.airflow.dag 导入 DbtDag。产物上传到对象存储:设置 AIRFLOW__COSMOS__REMOTE_TARGET_PATH 与 AIRFLOW__COSMOS__REMOTE_TARGET_PATH_CONN_ID,并从 cosmos.io 导入 upload_to_cloud_storage,在 operator_args 中设置 callback。dbt Docs 托管:Cosmos 在 Airflow UI 中提供 dbt docs,配置取决于 Airflow 主版本(两者使用不同 UI 插件系统,并非简单的单项目与多项目选择)。Airflow 2(FAB 插件)使用 DBT_DOCS_DIR(可加 DBT_DOCS_CONN_ID、DBT_DOCS_INDEX_FILE_NAME),作用域为单项目,Cosmos 1.4.0+;Airflow 3.1+(FastAPI)使用 DBT_DOCS_PROJECTS(JSON),作用域为一个或多个项目,Cosmos 1.11.0+。应按 Airflow 版本选择,而非按项目数量;单项目设置是 Airflow 2 的路径,Cosmos 未发布弃用通知,不应描述为 legacy 或 deprecated。

【相关技能】cosmos-dbt-fusion:用于 dbt Fusion 项目(非 dbt Core);authoring-dags:通用 DAG 编写模式;testing-dags:创建后测试 DAG。

本技能的目标用户是需要在 Airflow 中编排 dbt Core 项目的工程师,适用场景包括将现有 dbt 项目迁移为 Airflow DAG/TaskGroup、在已有 DAG 中嵌入 dbt 任务组,或直接使用 Cosmos 算子进行细粒度控制。限制方面:明确针对 dbt Core 而非 dbt Fusion,且面向 Cosmos 1.11+ 与 Airflow 3.x(Airflow 2.x 需按附录调整);AIRFLOW_ASYNC 模式仅支持 BigQuery。未在资料中提供的信息(如具体收费、平台限制、性能指标等)暂未提供。