Forum Discussion

rbrtr's avatar
rbrtr
Regular Visitor
1 year ago
Solved

Airflow DbtDag supported?

Hello, i tried the airflow dbt integration and I'm not able to run the dag and the dag is not imported into airflow either. It works when I switch from DbtDag to the default DAG definition. I followed the offical documentation. https://learn.microsoft.com/en-us/fabric/data-factory/apache-airflow-jobs-dbt-fabric


This definition doesnt work

 

 

 

import os
from pathlib import Path
from datetime import datetime
from cosmos import DbtDag, ProjectConfig, ProfileConfig, ExecutionConfig

DEFAULT_DBT_ROOT_PATH = Path(__file__).parent.parent / "dags" / "project_1"
DBT_ROOT_PATH = Path(os.getenv("DBT_ROOT_PATH", DEFAULT_DBT_ROOT_PATH))
profile_config = ProfileConfig(
     profile_name="default",
     target_name="fabric-dev",
     profiles_yml_filepath=DBT_ROOT_PATH / "profiles.yml",
)

# Define the default arguments for the DAG
dag_kd_test_project_1 = DbtDag(
     project_config=ProjectConfig(DBT_ROOT_PATH,),
     operator_args={"install_deps": True},
     profile_config=profile_config,
     schedule_interval="@daily",
     catchup=False,
     dag_id="test_project",
)

 

 

 

 

This definition works

 

 

 

from datetime import datetime
from airflow import DAG
from airflow.operators.bash import BashOperator

# Define the default arguments for the DAG
default_args = {
    'owner': 'airflow',
    'depends_on_past': False,
    'start_date': datetime(2023, 5, 1),
    'email_on_failure': False,
    'email_on_retry': False,
    'retries': 1
}

# Instantiate the DAG object
with DAG(
    'dags-helloworld.py',
    default_args=default_args,
    description='A simple Hello World DAG',
    schedule_interval="@daily",
    catchup=False
) as dag:

    # Define the tasks
    hello_task = BashOperator(
        task_id='hello_world_task',
        bash_command='echo "Hello, World!"'
    )

    # Set the task dependencies
    hello_task

 

 

 

 

  • Adding this import solved the problem

    from airflow import DAG

3 Replies

  • rbrtr's avatar
    rbrtr
    Regular Visitor

    Adding this import solved the problem

    from airflow import DAG
  • import os
    from pathlib import Path
    from datetime import datetime
    from cosmos import DbtDag, ProjectConfig, ProfileConfig, ExecutionConfig
    
    DEFAULT_DBT_ROOT_PATH = Path(__file__).parent.parent / "dags" / "project_1"
    DBT_ROOT_PATH = Path(os.getenv("DBT_ROOT_PATH", DEFAULT_DBT_ROOT_PATH))
    print(f"DBT_ROOT_PATH: {DBT_ROOT_PATH}")
    
    profile_config = ProfileConfig(
        profile_name="default",
        target_name="fabric-dev",
        profiles_yml_filepath=DBT_ROOT_PATH / "profiles.yml",
    )
    print(f"Profile Config: {profile_config}")
    
    # Define the default arguments for the DAG
    dag_kd_test_project_1 = DbtDag(
        project_config=ProjectConfig(DBT_ROOT_PATH),
        operator_args={"install_deps": True},
        profile_config=profile_config,
        schedule_interval="@daily",
        catchup=False,
        dag_id="test_project",
    )
    print(f"DAG: {dag_kd_test_project_1}")