cosmos-dbt-fusion

โดย astronomer

กำหนดค่า Astronomer Cosmos สำหรับโปรเจกต์ dbt Fusion บน Snowflake, Databricks, BigQuery หรือ Redshift ด้วยการทำงานในเครื่อง ต้องใช้ Cosmos 1.11.0+ ติดตั้ง dbt Fusion binary แยกต่างหากใน Airflow runtime และใช้ ExecutionMode.LOCAL พร้อมการเรียกใช้แบบ subprocess รองรับกลยุทธ์การแยกวิเคราะห์สามแบบ: dbt_manifest (เร็วที่สุดสำหรับโปรเจกต์ขนาดใหญ่), dbt_ls (สำหรับตัวเลือกที่ซับซ้อน) หรือ automatic (การตั้งค่าอย่างง่าย) ครอบคลุมการตั้งค่า ProfileConfig สำหรับการเชื่อมต่อ warehouse, 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-state-store
astronomer
Persists task and asset state across retries and DAG runs using Airflow 3.3's AIP-103 key/value stores (`task_state_store`, `asset_state_store`) and the…
creating-openlineage-extractors
astronomer
ตัวแยกข้อมูล OpenLineage แบบกำหนดเองสำหรับโอเปอเรเตอร์ Airflow ที่ไม่รองรับและสถานการณ์สายเลือดที่ซับซ้อน สองแนวทาง: เพิ่มเมธอด OpenLineage ลงในโอเปอเรเตอร์ที่คุณเป็นเจ้าของโดยตรง (แนะนำ) หรือสร้างตัวแยกข้อมูลแบบกำหนดเองสำหรับโอเปอเรเตอร์ของบุคคลที่สามที่คุณไม่สามารถแก้ไขได้ ตัวแยกข้อมูลจะสกัดกั้นการทำงานของโอเปอเรเตอร์ที่สามจุด: ก่อนการดำเนินการสำหรับสายเลือดแบบคงที่ หลังจากสำเร็จสำหรับเอาต์พุตที่กำหนดในรันไทม์ และหลังจากล้มเหลวสำหรับสายเลือดบางส่วน ลงทะเบียนตัวแยกข้อมูลผ่าน airflow.cfg หรือสภาพแวดล้อม...
debugging-dags
astronomer
การวิเคราะห์สาเหตุที่แท้จริงอย่างเป็นระบบและการแก้ไขสำหรับ Airflow DAGs ที่ล้มเหลว พร้อมขั้นตอนการตรวจสอบที่มีโครงสร้าง ชี้แนะผ่านกระบวนการวินิจฉัยสี่ขั้นตอน: ระบุความล้มเหลว ดึงรายละเอียดข้อผิดพลาด รวบรวมข้อมูลบริบท และส่งมอบขั้นตอนการแก้ไขที่สามารถดำเนินการได้ จัดหมวดหมู่ความล้มเหลวออกเป็นสี่ประเภท (ข้อมูล โค้ด โครงสร้างพื้นฐาน การพึ่งพา) เพื่อมุ่งเน้นการตรวจสอบและแนะนำการแก้ไขที่เหมาะสม ให้คำสั่ง CLI ที่พร้อมใช้งานสำหรับการดึงข้อมูลบันทึก การเปรียบเทียบการรัน การล้างงาน และ DAG...
delegating-to-otto
astronomer
Drives Astronomer's Otto agent (`astro otto`) as a delegated sub-agent for Airflow, dbt, and data-engineering work. Use when the user explicitly asks to "use…
deploying-airflow
astronomer
ปรับใช้ Airflow DAGs และโปรเจกต์ ใช้เมื่อผู้ใช้ต้องการปรับใช้โค้ด, ส่ง DAGs, ตั้งค่า CI/CD, ปรับใช้สู่ระบบผลิต หรือสอบถามเกี่ยวกับกลยุทธ์การปรับใช้…
deploying-go-sdk-bundles
astronomer
สร้าง, แพ็ก, และปรับใช้ชุดรวม Airflow Go SDK ที่คอมไพล์แล้ว เพื่อให้ ExecutableCoordinator สามารถรันได้ ใช้เมื่อผู้ใช้ต้องการคอมไพล์ชุดรวมงาน Go, ถาม…
testing-dags
astronomer
วงจรการทดสอบ-ดีบัก-แก้ไขแบบวนซ้ำสำหรับ Airflow DAGs พร้อมการวินิจฉัยข้อผิดพลาดอย่างครอบคลุม เริ่มต้นด้วย af runs trigger-wait <dag_id> เพื่อรัน DAG และรอให้เสร็จสมบูรณ์ ไม่จำเป็นต้องตรวจสอบก่อนเริ่มต้น เมื่อเกิดข้อผิดพลาด ให้ใช้ af runs diagnose เพื่อสรุปข้อผิดพลาดอย่างครอบคลุม และ af tasks logs เพื่อตรวจสอบรายละเอียดข้อผิดพลาดจากงานเฉพาะ รองรับการกำหนดค่าเอง การหมดเวลา และการลองใหม่ จัดการสถานการณ์สำเร็จ ล้มเหลว และหมดเวลาพร้อมการตีความผลลัพธ์ที่ชัดเจน มีการตรวจสอบความถูกต้องอย่างรวดเร็ว...
tracing-downstream-lineage
astronomer
ติดตามสายข้อมูลปลายน้ำเพื่อประเมินผลกระทบจากการเปลี่ยนแปลงก่อนปรับแก้ตารางหรือ DAG ระบุผู้บริโภคโดยตรงของตารางเป้าหมายหรือ DAG ผ่านการค้นหาในซอร์สโค้ด การขึ้นต่อกันของวิว และการเชื่อมต่อเครื่องมือ BI สร้างแผนผังการขึ้นต่อกันแบบสมบูรณ์ที่แสดงผลกระทบปลายน้ำทั้งหมด ตั้งแต่ตารางไปจนถึงแดชบอร์ดและโมเดล ML จัดหมวดหมู่การขึ้นต่อกันตามความสำคัญ (วิกฤต สูง ปานกลาง ต่ำ) เพื่อจัดลำดับความสำคัญในการสื่อสารกับผู้มีส่วนได้ส่วนเสียและการทดสอบ สร้างรายงานผลกระทบพร้อมการประเมินความเสี่ยง ผลกระทบที่ได้รับ...