astronomer-cosmos

Astronomer Cosmos allows you to run dbt Core projects as Apache Airflow DAGs and Task Groups with minimal code.

Key benefits include:

• Airflow Connection Integration: Run dbt projects using Airflow connections instead of managing separate dbt profiles.
• Virtual Environment Support: Native support for installing and running dbt in isolated virtual environments to prevent dependency conflicts with Airflow.
• Integrated Testing: Run dbt tests immediately after a model completes to catch data issues early.
• Data-Aware Scheduling: Leverage Airflow's data-aware scheduling to trigger model runs immediately after upstream data ingestion.
• Granular Task Control: Each dbt model is transformed into an individual Airflow task or task group, enabling per-model retries, alerting, and monitoring.

The jaffle_shop project is a fictional ecommerce store used as an example within this repository. It is a dbt project designed to transform raw data from an application database into analytics-ready models for customers and orders.

Note that this specific version has been modified from the original dbt project to demonstrate Cosmos features, specifically by adding tags to the models.

The dbt-date package provides several categories of macros for handling date logic:

• Date Dimension: Macros for generating date dimensions and base date sets.
• Calendar Date: Macros for date manipulation, timezone conversion, and extracting parts (day, month, week, etc.).
• Fiscal Date: Macros for handling fiscal periods based on custom year-end months and week starts.
• Utils: Helper macros for constructing date and datetime objects.

Cosmos performance tuning is divided into two primary dimensions:

1. DAG Parsing Speed: Affects how quickly DAGs appear and update in Apache Airflow. Optimization strategies include choosing the correct LoadMode, reducing DAG granularity, and skipping stale sources.
2. Task Execution Speed: Affects DAG run duration. Optimization strategies include choosing the right execution mode, sizing workers appropriately, and reducing per-task overhead.

Key optimization areas include:

• Rendering: Speeding up the parsing process.
• Execution: Speeding up the actual task runs.
• Memory: Reducing consumption during both parsing and execution.
• Caching: Leveraging cached dbt ls output, partial parse files, profiles, and YAML selectors.
• Invocation Mode: Choosing between running dbt as a library or as a subprocess.

The jaffle_shop project is a fictional ecommerce store used as an example within Cosmos. It demonstrates how to transform raw app database data into analytics-ready models for customers and orders.

This specific version of the project has been modified from the original dbt repository to showcase Cosmos-specific features:

• Model Tagging: Tags have been added to models to demonstrate how Cosmos can filter or group execution.
• Python Models: The standard SQL models (models/orders.sql and models/customers.sql) have been replaced with Python models (models/orders.py and models/customers.py) to demonstrate Cosmos's support for dbt Python models.
• Databricks Integration: The profiles.yml has been configured to interface with Databricks.

Cosmos is an open-source Python package that automatically converts dbt Core and dbt Fusion projects into Apache Airflow workflows (DAGs).

It functions by creating an interface between dbt and Airflow through two primary stages:

1. Parsing the dbt project: Cosmos parses your dbt project and translates it into an Airflow DAG. This stage is customized using ProjectConfig and RenderConfig to control how the dbt project is represented within Airflow.
2. Executing dbt commands: Cosmos executes the resulting DAG using configurations from ExecutionConfig and ProjectConfig. It uses the appropriate dbt adapter to run your SQL in your data warehouse, utilizing a connection defined in ProfileConfig.

This allows you to leverage dbt for data transformations while using Airflow's orchestration capabilities for scheduling and integration into end-to-end workflows.

Cosmos uses several configuration objects to manage the translation and execution of dbt projects within Airflow:

• ProjectConfig: Used during both the parsing and execution stages to define fundamental project settings.
• RenderConfig: Used during the parsing stage to customize how the dbt project is translated into an Airflow DAG.
• ExecutionConfig: Used during the execution stage to define how dbt commands are run.
• ProfileConfig: Defines the connection used to execute SQL in your data warehouse.

The RenderConfig parameters affect how dbt tests are executed depending on the TestBehavior:

• exclude: This parameter is passed to all test behaviors (AFTER_EACH, BUILD, and AFTER_ALL). For example, exclude=['resource_type:unit_test'] will ensure unit tests are excluded from both run and test tasks.
• select / selector: These are forwarded to test tasks only for TestBehavior.AFTER_ALL. They are NOT forwarded to per-model AFTER_EACH or BUILD tasks to avoid complex set semantics, as those tasks are already scoped to the specific model being tested.

from cosmos.airflow.dag import DbtDag
from cosmos.config import RenderConfig

jaffle_shop = DbtDag(
    render_config=RenderConfig(
        exclude=[
            "resource_type:unit_test"
        ],  # excluded from both the run and the test tasks
    )
)

The docker execution mode assumes you have a pre-built Docker image containing your dbt pipelines and a managed profiles.yml.

Key Considerations:

• Isolation: Provides better environment isolation than local or virtualenv modes.
• Maintenance: Requires more maintenance as you must ensure Docker containers have up-to-date files and manage secrets in multiple locations.
• Performance: Can be significantly slower than virtualenv because it may require building the Docker container before executing dbt commands.
• Docker-in-Docker: If your Airflow worker is already running in a container (e.g., in an Astro deployment), using ExecutionMode.DOCKER can cause issues related to running Docker inside Docker. It is generally advised against unless you have a specific requirement.

Interceptors (available in Cosmos 1.14+) are callables that run before Cosmos builds the dbt command for each task. Each interceptor receives (context, operator) and can modify operator.vars and operator.env. These modified values are then used during the dbt command execution.

To use them, pass them via operator_args in DbtDag or DbtTaskGroup:

operator_args={"interceptors": [...]}