cosmos-dbt-core

intent

turn a dbt Core project into an Airflow DAG, TaskGroup, or individual operators using Astronomer Cosmos. use this skill when you need to orchestrate dbt workflows inside Airflow, whether you're running a full project as a single DAG, embedding dbt tasks into an existing DAG, or fine-tuning execution with individual operators. does not cover dbt Fusion. before starting, verify your dbt engine is Core (not Fusion), confirm your warehouse type, check your Airflow version (2.x or 3.x), and decide on your execution pattern (standalone DAG, TaskGroup, or operators).

inputs

dbt project requirements:

• dbt engine: Core only (not Fusion)
• dbt project files: either local project directory or pre-built manifest.json + project_name
• warehouse type: Snowflake, BigQuery, Postgres, Redshift, Databricks, or other dbt-supported warehouse
• dbt profiles.yml or Airflow connection (ProfileMapping class preferred for managed secrets)

airflow environment:

• Airflow version: 2.x or 3.x (imports differ; see decision points)
• Cosmos version: 1.11+ (get latest from https://pypi.org/project/astronomer-cosmos/)
• execution environment: local Airflow, virtual environment, Kubernetes cluster, or container runtime

external connections:

• warehouse connection: stored in Airflow as conn_id (e.g., snowflake_default, bigquery_default). set up via Admin > Connections with credentials and warehouse params
• dbt cloud API (optional): if using manifest_path, dbt Cloud API is not required; if doing remote operations, set DBT_CLOUD_API_TOKEN env var

files and paths:

• dbt_project_path: full path to dbt project root (contains dbt_project.yml)
• manifest_path (optional): path to target/manifest.json if using manifest-only parsing
• profiles.yml path (optional): if using file-based profiles instead of Airflow connections

procedure

1. choose your assembly pattern.

• DbtDag: wrap entire dbt project as a standalone DAG. use when dbt is your primary workflow.
• DbtTaskGroup: embed dbt project inside an existing Airflow DAG. use when dbt is one part of a larger pipeline.
• individual operators: use DbtRunOperator, DbtTestOperator, etc. for granular control. use when you need custom orchestration between dbt tasks.

2. verify dbt project and warehouse connection. input: dbt_project_path or manifest_path, warehouse type (Snowflake, BigQuery, Postgres, etc.)

• confirm dbt_project.yml exists at dbt_project_path or manifest.json exists at manifest_path.
• confirm warehouse credentials are stored in Airflow as a connection (Conn ID visible in Admin > Connections).
• run dbt debug locally to confirm connectivity before wiring into Cosmos. output: confirmed path, confirmed conn_id, confirmed warehouse connectivity.

3. configure ProjectConfig. input: dbt_project_path or manifest_path + project_name.

• if using local files: set dbt_project_path = "/path/to/dbt/project", manifest_path = None.
• if using manifest only (faster, no file access needed): set manifest_path = "target/manifest.json", project_name = "my_dbt_project", dbt_project_path = None. output: ProjectConfig object with paths set correctly.

4. choose execution mode. input: isolation level needed, performance constraint, available resources. examine these eight modes:

• WATCHER: fastest. parses dbt_manifest.json. no subprocess. best for projects with stable selectors.
• LOCAL: default. runs dbt in a subprocess on the same machine. good for development.
• VIRTUALENV: isolates dbt runtime in a venv. good for dependency conflicts.
• KUBERNETES: runs each task in a K8s pod. best for scalability and resource isolation.
• AIRFLOW_ASYNC: async execution. reduces task footprint.
• DOCKER: runs dbt in a Docker container. requires Docker engine.
• DBT_LS: parses via dbt ls command. slower but handles complex selectors.
• DBT_LS_FILE: writes dbt ls output to file before parsing. good for large projects. output: selected execution_mode (e.g., "local", "virtualenv", "kubernetes").

5. choose parsing strategy. input: project size, selector complexity, speed vs. flexibility tradeoff.

• dbt_manifest: reads manifest.json directly. fastest. use if manifest is always fresh. not compatible with dynamic selectors like --select state:new.
• dbt_ls: runs dbt ls command. slower. handles all selectors (state:, tag:, path:, etc.).
• dbt_ls_file: writes dbt ls output to file. good middle ground for large projects.
• automatic: Cosmos picks the best strategy. safe default. output: selected parsing_strategy (e.g., "dbt_manifest", "dbt_ls", "automatic").

6. configure warehouse connection via ProfileMapping or profiles.yml. input: warehouse type, credentials, conn_id. approach a: use ProfileMapping (recommended).

• import ProfileMapping subclass for your warehouse (SnowflakeProfileMapping, BigQueryProfileMapping, PostgresProfileMapping, etc.).
• pass conn_id = "snowflake_default" (or your Airflow conn_id).
• Cosmos reads credentials from Airflow connection at runtime. output: ProfileMapping object linked to conn_id.

approach b: use existing profiles.yml (legacy).

• set profiles_yml_filepath = "/home/user/.dbt/profiles.yml".
• confirm dbt profile name matches project_name in dbt_project.yml. output: profiles.yml parsed by Cosmos.

7. configure test behavior. input: testing strategy, when to run tests.

• AFTER_EACH: run tests after each model. catches issues early. slower.
• AFTER_ALL: run tests once after all models. faster. defers error detection.
• BUILD: run dbt build (models + tests in dependency order). simplest.
• NONE: skip tests. use only in CI/development branches. output: test_behavior set (e.g., "after_each").

8. instantiate and deploy. input: all configs from steps 1-7.

• for DbtDag: instantiate DbtDag(dag_id="my_dbt_project", project_config=pc, ...).
• for DbtTaskGroup: instantiate DbtTaskGroup(dag_id="parent_dag", task_id="dbt_tasks", project_config=pc, ...).
• for operators: instantiate DbtRunOperator, DbtTestOperator individually in your DAG. output: DAG or TaskGroup registered in Airflow, visible in UI.

9. trigger and monitor. input: deployed DAG/TaskGroup.

• trigger manually or via schedule_interval.
• monitor in Airflow UI: watch task logs, check dbt artifacts (manifest, run_results.json) in Cosmos-generated folders. output: successful dbt runs logged in Airflow, dbt test results visible in task logs.

decision points

if airflow 2.x (not 3.x):

• use import cosmos from astronomer.cosmos (not from airflow.cosmos).
• check Cosmos 1.11+ compatibility matrix for Airflow 2.x versions.
• else: Airflow 3.x imports from airflow.cosmos natively.

if dbt engine is Fusion (not Core):

• stop. use cosmos-dbt-fusion skill instead.
• else: proceed with cosmos-dbt-core.

if manifest.json is not available or stale:

• switch parsing_strategy from "dbt_manifest" to "dbt_ls" or "automatic".
• this triggers dbt ls at runtime, slower but always fresh.
• else: use "dbt_manifest" for speed.

if execution environment is bare Airflow (no isolation):

• use execution_mode = "local". dbt subprocess shares Python environment.
• else if isolation needed: use "virtualenv", "docker", or "kubernetes".

if warehouse credentials are in profiles.yml file:

• use profiles_yml_filepath approach.
• else if credentials in Airflow connection: use ProfileMapping approach (preferred).

if test_behavior is not set and you want dbt tests to run:

• default is "after_all". if you need earlier detection, set "after_each" (slower).
• else if no tests needed: set "none".

if dbt project uses custom packages or macros:

• ensure dbt packages.yml is in dbt_project_path.
• Cosmos runs dbt deps at task startup, no extra config needed.
• else: no action needed.

if dbt selector is dynamic (e.g., state:new, state:modified):

• parsing_strategy must be "dbt_ls" or "automatic", not "dbt_manifest".
• "dbt_manifest" will fail silently or skip dynamic selectors.
• else: "dbt_manifest" is fine.

if Cosmos job times out or tasks fail with "connection refused":

• check warehouse connection: run dbt debug in terminal to confirm.
• check Airflow conn_id exists and matches ProfileMapping conn_id param.
• check firewall / network rules allow Airflow worker to reach warehouse.
• else: escalate to dbt or warehouse logs.

output contract

successful deployment produces:

• Airflow DAG or TaskGroup registered in Airflow UI (visible under DAGs tab).
• Cosmos-generated task folder at airflow_home/dbt/{dag_id}/{task_id} containing:
  • target/manifest.json (parsed dbt metadata)
  • target/run_results.json (task execution results)
  • logs/ folder (dbt command logs)
• each task in the DAG/TaskGroup maps to a dbt node (model, test, snapshot, seed).
• task names follow pattern: {project_name}.{node_name} (e.g., "my_dbt_project.stg_customers").

if using DbtDag:

• single DAG instance with all dbt nodes