cosmos-dbt-fusion

Настройка Astronomer Cosmos для проектов dbt Fusion на Snowflake, Databricks, BigQuery или Redshift с локальным выполнением. Требуется Cosmos 1.11.0+, отдельная установка бинарного файла dbt Fusion в среде выполнения Airflow и ExecutionMode.LOCAL с вызовом подпроцесса. Поддерживает три стратегии парсинга: dbt_manifest (самая быстрая для крупных проектов), dbt_ls (для сложных селекторов) или автоматическая (для простых конфигураций). Охватывает настройку ProfileConfig для подключения к хранилищам, ProjectConfig для путей к проектам dbt и...

npx skills add https://github.com/astronomer/agents --skill cosmos-dbt-fusion

Cosmos + dbt Fusion: Implementation Checklist

Execute steps in order. This skill covers Fusion-specific constraints only.

Version note: dbt Fusion support was introduced in Cosmos 1.11.0. Requires Cosmos ≥1.11.

Reference: See reference/cosmos-config.md for ProfileConfig, operator_args, and Airflow 3 compatibility details.

Before starting, confirm: (1) dbt engine = Fusion (not Core → use cosmos-dbt-core), (2) warehouse = Snowflake, Databricks, Bigquery and Redshift only.

Fusion-Specific Constraints

ConstraintDetails
No asyncAIRFLOW_ASYNC not supported
No virtualenvFusion is a binary, not a Python package
Warehouse supportSnowflake, Databricks, Bigquery and Redshift support while in preview

1. Confirm Cosmos Version

CRITICAL: Cosmos 1.11.0 introduced dbt Fusion compatibility.

# Check installed version
pip show astronomer-cosmos

# Install/upgrade if needed
pip install "astronomer-cosmos>=1.11.0"

Validate: pip show astronomer-cosmos reports version ≥ 1.11.0


2. Install the dbt Fusion Binary (REQUIRED)

dbt Fusion is NOT bundled with Cosmos or dbt Core. Install it into the Airflow runtime/image.

Determine where to install the Fusion binary (Dockerfile / base image / runtime).

Example Dockerfile Install

USER root
RUN apt-get update && apt-get install -y curl
ENV SHELL=/bin/bash
RUN curl -fsSL https://public.cdn.getdbt.com/fs/install/install.sh | sh -s -- --update
USER astro

Common Install Paths

EnvironmentTypical path
Astro Runtime/home/astro/.local/bin/dbt
System-wide/usr/local/bin/dbt

Validate: The dbt binary exists at the chosen path and dbt --version succeeds.


3. Choose Parsing Strategy (RenderConfig)

Parsing strategy is the same as dbt Core. Pick ONE:

Load modeWhen to useRequired inputs
dbt_manifestLarge projects; fastest parsingProjectConfig.manifest_path
dbt_lsComplex selectors; need dbt-native selectionFusion binary accessible to scheduler
automaticSimple setups; let Cosmos pick(none)
from cosmos import RenderConfig, LoadMode

_render_config = RenderConfig(
    load_method=LoadMode.AUTOMATIC,  # or DBT_MANIFEST, DBT_LS
)

4. Configure Warehouse Connection (ProfileConfig)

Reference: See reference/cosmos-config.md for full ProfileConfig options and examples.

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",
    ),
)

5. Configure ExecutionConfig (LOCAL Only)

CRITICAL: dbt Fusion with Cosmos requires ExecutionMode.LOCAL with dbt_executable_path pointing to the Fusion binary.

from cosmos import ExecutionConfig
from cosmos.constants import InvocationMode

_execution_config = ExecutionConfig(
    invocation_mode=InvocationMode.SUBPROCESS,
    dbt_executable_path="/home/astro/.local/bin/dbt",  # REQUIRED: path to Fusion binary
    # execution_mode is LOCAL by default - do not change
)

6. Configure Project (ProjectConfig)

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
    # install_dbt_deps=False,  # if deps precomputed in CI
)

7. Assemble DAG / TaskGroup

Option A: DbtDag (Standalone)

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(
    dbt_executable_path="/home/astro/.local/bin/dbt",  # Fusion binary
)

_render_config = RenderConfig()

my_fusion_dag = DbtDag(
    dag_id="my_fusion_cosmos_dag",
    project_config=_project_config,
    profile_config=_profile_config,
    execution_config=_execution_config,
    render_config=_render_config,
    start_date=datetime(2025, 1, 1),
    schedule="@daily",
)

Option B: DbtTaskGroup (Inside Existing 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
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(dbt_executable_path="/home/astro/.local/bin/dbt")

@dag(start_date=datetime(2025, 1, 1), schedule="@daily")
def my_dag():
    @task
    def pre_dbt():
        return "some_value"

    dbt = DbtTaskGroup(
        group_id="dbt_fusion_project",
        project_config=_project_config,
        profile_config=_profile_config,
        execution_config=_execution_config,
    )

    @task
    def post_dbt():
        pass

    chain(pre_dbt(), dbt, post_dbt())

my_dag()

8. Final Validation

Before finalizing, verify:

  • Cosmos version: ≥1.11.0
  • Fusion binary installed: Path exists and is executable
  • Warehouse supported: Snowflake, Databricks, Bigquery or Redshift only
  • Secrets handling: Airflow connections or env vars, NOT plaintext

Troubleshooting

If user reports dbt Core regressions after enabling Fusion:

AIRFLOW__COSMOS__PRE_DBT_FUSION=1

User Must Test

  • The DAG parses in the Airflow UI (no import/parse-time errors)
  • A manual run succeeds against the target warehouse (at least one model)

Reference


Related Skills

  • cosmos-dbt-core: For dbt Core projects (not Fusion)
  • authoring-dags: General DAG authoring patterns
  • testing-dags: Testing DAGs after creation

Больше skills от astronomer

airflow
astronomer
Запрос, управление и устранение неполадок DAG, запусков, задач и системной конфигурации Apache Airflow. Поддерживает более 30 команд для проверки DAG, управления запусками, ведения журналов задач, запросов конфигурации и прямого доступа к REST API. Управление несколькими экземплярами Airflow с постоянной конфигурацией; автоматическое обнаружение локальных и Astro развертываний. Синхронный (с ожиданием завершения) или асинхронный запуск DAG, диагностика сбоев, очистка запусков для повторного выполнения, доступ к журналам задач с фильтрацией по повторным попыткам и индексу карты. Вывод...
official
airflow-hitl
astronomer
Шлюзы утверждения человеком, ввод форм и ветвление в DAG Airflow с использованием отложенных операторов. Четыре типа операторов: ApprovalOperator для решений утвердить/отклонить, HITLOperator для выбора нескольких вариантов с формами, HITLBranchOperator для маршрутизации задач на основе решений человека и HITLEntryOperator для сбора данных из форм. Все операторы являются отложенными, освобождая слоты рабочих узлов в ожидании ответа человека через вкладку Required Actions в интерфейсе Airflow или REST API. Поддерживает дополнительные функции, включая пользовательские...
official
airflow-plugins
astronomer
Создавайте плагины для Airflow 3.1+, которые встраивают приложения FastAPI, пользовательские страницы интерфейса, компоненты React, промежуточное ПО, макросы и ссылки на операторы непосредственно в интерфейс Airflow. Используйте…
official
analyzing-data
astronomer
Запрашивайте данные из вашего хранилища данных для ответа на бизнес-вопросы с использованием кэшированных шаблонов и сопоставлений понятий. Поддерживает поиск по шаблонам и кэширование для повторяющихся типов вопросов с записью результатов для улучшения будущих запросов. Включает кэш сопоставлений понятий и таблиц, а также обнаружение схем таблиц через INFORMATION_SCHEMA или grep кодовой базы. Предоставляет функции ядра run_sql() и run_sql_pandas(), возвращающие DataFrames Polars или Pandas для анализа. Команды CLI для управления кэшами понятий, шаблонов и таблиц, а также...
official
annotating-task-lineage
astronomer
Аннотирование задач Airflow с помощью data lineage с использованием inlets и outlets. Поддерживает объекты Dataset OpenLineage, Airflow Assets и Airflow Datasets для определения входных и выходных данных в базах данных, хранилищах данных и облачных хранилищах. Используется как запасной вариант, когда операторам не хватает встроенных экстракторов OpenLineage; следует четырехуровневой системе приоритетов, где пользовательские экстракторы и методы OpenLineage имеют приоритет. Включает вспомогательные функции для именования наборов данных для Snowflake, BigQuery, S3 и PostgreSQL для обеспечения согласованности...
official
authoring-dags
astronomer
Пошаговый процесс создания DAG Apache Airflow с интеграцией валидации и тестирования. Структурированный шестифазный подход: обнаружение среды и существующих шаблонов, планирование структуры DAG, реализация с соблюдением лучших практик, валидация с помощью команд af CLI, тестирование с согласия пользователя и итеративное исправление. Команды CLI для обнаружения (af config connections, af config providers, af dags list) и валидации (af dags errors, af dags get, af dags explore) обеспечивают немедленную обратную связь по DAG...
official
blueprint
astronomer
Определяйте переиспользуемые шаблоны групп задач Airflow с валидацией Pydantic и составляйте DAG из YAML. Используйте при создании шаблонов blueprint, составлении DAG из…
official
checking-freshness
astronomer
Проверяет свежесть данных, сравнивая временные метки таблиц и шаблоны обновлений со шкалой устаревания. Определяет столбцы с временными метками, используя распространённые шаблоны именования ETL (_loaded_at, _updated_at, created_at и т.д.), и запрашивает их максимальные значения для определения возраста. Классифицирует данные по четырём статусам свежести: Свежие (< 4 часов), Устаревшие (4–24 часа), Очень устаревшие (> 24 часов) или Неизвестно (временная метка не найдена). Предоставляет SQL-шаблоны для проверки времени последнего обновления и тенденций количества строк за последние дни, чтобы...
official