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
astronomer
สอบถาม จัดการ และแก้ไขปัญหา DAGs, การรัน, งาน และการกำหนดค่าระบบของ Apache Airflow รองรับคำสั่งมากกว่า 30 คำสั่งสำหรับการตรวจสอบ DAG, การจัดการการรัน, การบันทึกงาน, การสอบถามการกำหนดค่า และการเข้าถึง REST API โดยตรง จัดการอินสแตนซ์ Airflow หลายตัวพร้อมการกำหนดค่าถาวร ค้นหาการปรับใช้ในเครื่องและ Astro โดยอัตโนมัติ เรียกใช้ DAG แบบซิงโครนัส (รอให้เสร็จ) หรือแบบอะซิงโครนัส วินิจฉัยข้อผิดพลาด ล้างการรันเพื่อลองใหม่ และเข้าถึงบันทึกงานพร้อมการกรอง retry/map-index ผลลัพธ์...
official
airflow-hitl
astronomer
ประตูการอนุมัติของมนุษย์, การป้อนข้อมูลฟอร์ม, และการแตกกิ่งใน Airflow DAGs โดยใช้ตัวดำเนินการที่สามารถเลื่อนได้ ตัวดำเนินการสี่ประเภท: ApprovalOperator สำหรับการตัดสินใจอนุมัติ/ปฏิเสธ, HITLOperator สำหรับการเลือกหลายตัวเลือกพร้อมฟอร์ม, HITLBranchOperator สำหรับการกำหนดเส้นทางงานที่ขับเคลื่อนโดยมนุษย์, และ HITLEntryOperator สำหรับการรวบรวมข้อมูลฟอร์ม ตัวดำเนินการทั้งหมดสามารถเลื่อนได้ โดยปล่อยช่อง worker ขณะรอการตอบสนองจากมนุษย์ผ่านแท็บ Required Actions ของ Airflow UI หรือ REST API รองรับคุณสมบัติเสริมรวมถึงแบบกำหนดเอง...
official
airflow-plugins
astronomer
สร้างปลั๊กอิน Airflow 3.1+ ที่ฝังแอป FastAPI, หน้า UI แบบกำหนดเอง, คอมโพเนนต์ React, มิดเดิลแวร์, มาโคร และลิงก์โอเปอเรเตอร์ลงใน UI ของ Airflow โดยตรง ใช้…
official
analyzing-data
astronomer
สอบถามคลังข้อมูลของคุณเพื่อตอบคำถามทางธุรกิจด้วยรูปแบบที่แคชไว้และการแมปแนวคิด รองรับการค้นหารูปแบบและการแคชสำหรับประเภทคำถามที่เกิดซ้ำ พร้อมบันทึกผลลัพธ์เพื่อปรับปรุงการสอบถามในอนาคต รวมถึงแคชการแมปแนวคิดไปยังตารางและการค้นพบสคีมาตารางผ่าน INFORMATION_SCHEMA หรือการ grep โค้ดเบส มีฟังก์ชันเคอร์เนล run_sql() และ run_sql_pandas() ที่ส่งคืน Polars หรือ Pandas DataFrames สำหรับการวิเคราะห์ คำสั่ง CLI สำหรับจัดการแคชแนวคิด รูปแบบ และตาราง รวมถึง...
official
annotating-task-lineage
astronomer
ใส่คำอธิบาย Airflow tasks ด้วย data lineage โดยใช้ inlets และ outlets รองรับ OpenLineage Dataset objects, Airflow Assets และ Airflow Datasets สำหรับกำหนด inputs และ outputs ครอบคลุมฐานข้อมูล, data warehouses และ cloud storage ใช้เป็นทางเลือกสำรองเมื่อ operators ไม่มี OpenLineage extractors ในตัว; ทำงานตามระบบลำดับความสำคัญสี่ระดับที่ custom extractors และ OpenLineage methods มีสิทธิ์優先 รวมถึงตัวช่วยตั้งชื่อ dataset สำหรับ Snowflake, BigQuery, S3 และ PostgreSQL เพื่อให้มั่นใจถึงความสอดคล้อง...
official
authoring-dags
astronomer
เวิร์กโฟลว์แบบมีคำแนะนำสำหรับสร้าง Apache Airflow DAGs พร้อมการตรวจสอบความถูกต้องและการผสานการทดสอบ แนวทางแบบหกขั้นตอน: ค้นพบสภาพแวดล้อมและรูปแบบที่มีอยู่ วางแผนโครงสร้าง DAG ดำเนินการตามแนวทางปฏิบัติที่ดีที่สุด ตรวจสอบความถูกต้องด้วยคำสั่ง CLI ของ af ทดสอบโดยได้รับความยินยอมจากผู้ใช้ และปรับปรุงแก้ไขซ้ำ คำสั่ง 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