Astronomer Cosmos

Warnings

• breaking Cosmos v1.14.0 (alpha releases) drops support for Apache Airflow versions earlier than 2.9.
  Fix: Upgrade your Airflow environment to version 2.9 or higher before upgrading to Cosmos 1.14.0 or later.
• gotcha Airflow's `DagBag` import timeout can occur with large dbt projects due to slow manifest parsing, especially when using `ExecutionMode.LOCAL` without a pre-generated `manifest.json`.
  Fix: Increase the `core.dagbag_import_timeout` Airflow configuration. Consider refactoring Cosmos DAGs to use a pre-generated `manifest.json` file, or explicitly setting `LoadMode.DBT_MANIFEST` and pre-generating the manifest during image build or deployment.
• gotcha Dependency conflicts between `dbt-core`/dbt adapters and Airflow packages are common.
  Fix: Astronomer Cosmos strongly recommends installing dbt and its adapters within a separate Python virtual environment, distinct from the one where Airflow is installed. The `dbt_executable_path` in `ExecutionConfig` should then point to this isolated virtual environment's dbt executable.
• gotcha Cosmos's default dbt test behavior (`TestBehavior.AFTER_EACH`) runs tests immediately after each model, which differs from dbt Core's default of running all tests after all models have completed. This 'fail-fast' approach can sometimes lead to unexpected failures, especially for tests that depend on multiple upstream models.
  Fix: If this behavior is not desired, configure `RenderConfig` to use `TestBehavior.AFTER_ALL`. For tests with multiple parents, consider setting `should_detach_multiple_parents_tests=True` in `RenderConfig` (introduced in Cosmos 1.8.2) to create separate test tasks for better isolation.
• gotcha Intermittent `FileNotFoundError: [Errno 2] No such file or directory` errors for `/tmp` files can occur during dbt model run tasks, particularly with `ExecutionMode.LOCAL`.
  Fix: This issue is often related to concurrent dbt processes and temporary file management. Ensure proper temporary directory handling, explicitly configure `TMPDIR` environment variables if needed, and increase task retries (e.g., `retries=2`) as re-runs often succeed.
• gotcha When incorporating dbt snapshots, ensure they are run in a separate `DbtTaskGroup` from dbt models to guarantee the correct execution order.
  Fix: Create distinct `DbtTaskGroup` instances for your snapshots and models, and chain them in your Airflow DAG to enforce the desired sequence (e.g., snapshots before models).

Install

• pip install astronomer-cosmos Install core library
• pip install 'astronomer-cosmos[dbt-<adapter>]' Install with dbt adapter (e.g., dbt-postgres)

Imports

• DbtDag

  from cosmos import DbtDag

• DbtTaskGroup

  from cosmos import DbtTaskGroup

• ProjectConfig

  from cosmos import ProjectConfig

• ProfileConfig

  from cosmos import ProfileConfig

• ExecutionConfig

  from cosmos import ExecutionConfig

• PostgresUserPasswordProfileMapping

  from cosmos.profiles import PostgresUserPasswordProfileMapping

  Import specific profile mappings from `cosmos.profiles` based on your dbt adapter and connection type (e.g., `SnowflakeUserPasswordProfileMapping`).

Quickstart

This quickstart demonstrates how to create an Airflow DAG that orchestrates a dbt project using Astronomer Cosmos's `DbtDag` class. It configures a dbt project path, maps an Airflow connection to a dbt profile, and sets an execution environment for dbt. Replace placeholder values like `AIRFLOW_DB_CONN_ID` and `DBT_EXECUTABLE_PATH` with your actual environment variables or paths. Ensure your dbt project (e.g., 'jaffle_shop') is accessible from your Airflow environment.

import os
from datetime import datetime
from pathlib import Path

from airflow.models.dag import DAG
from cosmos import DbtDag, ProjectConfig, ProfileConfig, ExecutionConfig
from cosmos.profiles import PostgresUserPasswordProfileMapping

# Define path to your dbt project relative to the DAG file
# For this example, assume a dbt project named 'jaffle_shop' exists in 'dags/dbt/'
# Example structure: dags/my_cosmos_dag.py, dags/dbt/jaffle_shop/dbt_project.yml
DEFAULT_DBT_PROJECT_PATH = Path(__file__).parent / "dbt" / "jaffle_shop"

# Configure your dbt profile to use an Airflow connection
# Replace 'your_airflow_postgres_conn_id' with an actual Airflow connection ID
# and adjust profile_args as needed for your database schema.
profile_config = ProfileConfig(
    profile_name="default", # Corresponds to the profile name in your dbt_project.yml
    target_name="dev",      # Corresponds to the target name in your dbt_project.yml
    profile_mapping=PostgresUserPasswordProfileMapping(
        conn_id=os.environ.get('AIRFLOW_DB_CONN_ID', 'airflow_db'), # Airflow connection ID
        profile_args={"schema": "public"}, # Example schema
    ),
)

# Configure dbt execution, e.g., to run dbt in a virtual environment.
# Ensure 'dbt_executable_path' points to your dbt virtual environment's executable.
# If dbt is installed in the same environment as Airflow, this might be omitted or set to 'dbt'.
execution_config = ExecutionConfig(
    dbt_executable_path=os.environ.get('DBT_EXECUTABLE_PATH', '/opt/airflow/dbt_venv/bin/dbt'),
)

with DAG(
    dag_id="cosmos_dbt_example_dag",
    schedule_interval="@daily",
    start_date=datetime(2023, 1, 1),
    catchup=False,
    tags=["dbt", "cosmos", "example"],
    default_args={
        "retries": 2, # Recommended for dbt tasks
    },
) as dag:
    dbt_project = DbtDag(
        project_config=ProjectConfig(
            DEFAULT_DBT_PROJECT_PATH,
        ),
        profile_config=profile_config,
        execution_config=execution_config,
        operator_args={
            "install_deps": True, # Install dbt dependencies before run
        },
    )

    # The dbt_project object automatically creates Airflow tasks for your dbt models,
    # which can be observed in the Airflow UI with their lineage.
    # You can add upstream/downstream Airflow tasks as needed.
    # For example, a simple PythonOperator before or after the dbt_project.