Orchestrating dbt with Airflow - Mini Project: Build & Apply

Orchestrating dbt with Airflow

📖 Scenario: You work as a data analyst in a company that uses dbt (data build tool) to transform raw data into clean, usable tables. To automate these transformations, your team uses Apache Airflow, a tool that schedules and runs workflows.Today, you will create a simple Airflow DAG (Directed Acyclic Graph) to run dbt commands automatically. This will help your team keep data fresh without manual work.

🎯 Goal: Build an Airflow DAG that runs dbt commands to build and test your data models. You will create the DAG, configure the dbt commands, and print the results.

📋 What You'll Learn

Create a Python dictionary called default_args with Airflow DAG default settings.

Create a variable called dbt_run_command with the dbt run shell command as a string.

Create a Python function called run_dbt that prints the dbt run command.

Create an Airflow DAG named dbt_dag that uses the run_dbt function as a task.

Print the task id of the dbt run task.

💡 Why This Matters

🌍 Real World

Automating dbt runs with Airflow helps data teams keep data models updated daily without manual intervention.

💼 Career

Data engineers and analysts often use Airflow to schedule and monitor dbt workflows in production data pipelines.

1

Set up Airflow DAG default arguments

Create a Python dictionary called default_args with these exact entries: 'owner': 'airflow', 'start_date': datetime(2024, 1, 1), and 'retries': 1. Import datetime from the datetime module first.

dbt

from datetime import datetime

# Create default_args dictionary with owner, start_date, and retries
# Your code here

Hint

Use default_args = {'owner': 'airflow', 'start_date': datetime(2024, 1, 1), 'retries': 1}.

2

Create the dbt run command string

Create a variable called dbt_run_command and set it to the string 'dbt run --profiles-dir ./profiles' exactly.

dbt

from datetime import datetime

default_args = {
    'owner': 'airflow',
    'start_date': datetime(2024, 1, 1),
    'retries': 1
}

# Create dbt_run_command string with the exact dbt run command
# Your code here

Hint

Set dbt_run_command = 'dbt run --profiles-dir ./profiles'.

3

Define the dbt run function and Airflow DAG

Import DAG from airflow and PythonOperator from airflow.operators.python. Define a function called run_dbt that prints the dbt_run_command. Then create an Airflow DAG called dbt_dag with default_args and schedule_interval='@daily'. Inside the DAG context, create a PythonOperator task called dbt_run_task that runs the run_dbt function.

dbt

from datetime import datetime
from airflow import DAG
from airflow.operators.python import PythonOperator

default_args = {
    'owner': 'airflow',
    'start_date': datetime(2024, 1, 1),
    'retries': 1
}

dbt_run_command = 'dbt run --profiles-dir ./profiles'

# Define run_dbt function that prints dbt_run_command
# Create DAG named dbt_dag with default_args and schedule_interval '@daily'
# Inside DAG, create PythonOperator task dbt_run_task that calls run_dbt
# Your code here

Hint

Use with DAG('dbt_dag', default_args=default_args, schedule_interval='@daily') as dag: and create dbt_run_task = PythonOperator(task_id='dbt_run', python_callable=run_dbt).

4

Print the task id of the dbt run task

Write a print statement to display the task_id of the dbt_run_task.

dbt

from datetime import datetime
from airflow import DAG
from airflow.operators.python import PythonOperator

default_args = {
    'owner': 'airflow',
    'start_date': datetime(2024, 1, 1),
    'retries': 1
}

dbt_run_command = 'dbt run --profiles-dir ./profiles'

def run_dbt():
    print(dbt_run_command)

with DAG('dbt_dag', default_args=default_args, schedule_interval='@daily') as dag:
    dbt_run_task = PythonOperator(
        task_id='dbt_run',
        python_callable=run_dbt
    )

# Print the task_id of dbt_run_task
# Your code here

Hint

Use print(dbt_run_task.task_id) to show the task id.

Practice

1. What is the main purpose of using Airflow with dbt in data workflows?

easy

A. To automate and schedule dbt commands in a specific order

B. To replace dbt's SQL models with Python scripts

C. To visualize dbt models as charts

D. To manually run dbt commands from the command line

Solution

1. Step 1: Understand Airflow's role

   Airflow is a tool to automate and schedule tasks, including dbt commands.

2. Step 2: Connect Airflow with dbt

   Using Airflow, you can run dbt commands automatically and in order without manual intervention.

3. Final Answer:

   To automate and schedule dbt commands in a specific order -> Option A

4. Quick Check:

   Airflow automates dbt runs = C [OK]

Hint: Airflow schedules tasks; dbt runs models. Together automate workflows [OK]

Common Mistakes:

• Thinking Airflow replaces dbt's SQL modeling
• Confusing manual and automated runs
• Assuming Airflow visualizes dbt models

2. Which Airflow operator is commonly used to run dbt commands inside a DAG?

easy

A. BashOperator

B. PythonOperator

C. HttpOperator

D. EmailOperator

Solution

1. Step 1: Identify how dbt commands run

   dbt commands run in the shell or terminal, so Airflow needs an operator that runs shell commands.

2. Step 2: Match operator to shell commands

   BashOperator runs bash shell commands, perfect for running dbt CLI commands inside Airflow.

3. Final Answer:

   BashOperator -> Option A

4. Quick Check:

   BashOperator runs shell commands = A [OK]

Hint: Use BashOperator for shell commands like dbt runs [OK]

Common Mistakes:

• Choosing PythonOperator which runs Python code, not shell commands
• Confusing HttpOperator with web requests
• Thinking EmailOperator runs commands

3. Given this Airflow DAG snippet, what is the order of task execution?

dbt_run = BashOperator(task_id='dbt_run', bash_command='dbt run')
dbt_test = BashOperator(task_id='dbt_test', bash_command='dbt test')
dbt_run >> dbt_test

medium

A. dbt_run and dbt_test run in parallel

B. dbt_run runs first, then dbt_test

C. dbt_test runs first, then dbt_run

D. Only dbt_test runs

Solution

1. Step 1: Understand the dependency operator

   The '>>' operator in Airflow means the left task runs before the right task.

2. Step 2: Apply to given tasks

   dbt_run >> dbt_test means dbt_run runs first, then dbt_test runs after it finishes.

3. Final Answer:

   dbt_run runs first, then dbt_test -> Option B

4. Quick Check:

   dbt_run >> dbt_test means run then test = A [OK]

Hint: >> means left task runs before right task [OK]

Common Mistakes:

• Thinking >> means parallel execution
• Reversing the order of tasks
• Assuming only one task runs

4. Identify the error in this Airflow task setup for running dbt tests:

dbt_test = BashOperator(task_id='dbt_test', bash_command='dbt test')
dbt_test.set_upstream('dbt_run')

medium

A. bash_command should be a list, not a string

B. task_id cannot be 'dbt_test'

C. set_upstream expects a task object, not a string

D. BashOperator cannot run dbt commands

Solution

1. Step 1: Check set_upstream usage

   set_upstream expects another task object, not a string task id.

2. Step 2: Identify correct usage

   It should be dbt_test.set_upstream(dbt_run) where dbt_run is the task object, not 'dbt_run' string.

3. Final Answer:

   set_upstream expects a task object, not a string -> Option C

4. Quick Check:

   set_upstream needs task object = B [OK]

Hint: set_upstream needs task object, not string name [OK]

Common Mistakes:

• Passing string instead of task object to set_upstream
• Thinking bash_command must be a list
• Believing task_id has naming restrictions
• Assuming BashOperator can't run dbt

5. You want to create an Airflow DAG that runs dbt run, then dbt test, and finally sends an email notification only if tests pass. Which sequence correctly sets task dependencies?

hard

A. dbt_run >> email_notify >> dbt_test

B. email_notify >> dbt_run >> dbt_test

C. dbt_test >> dbt_run >> email_notify

D. dbt_run >> dbt_test >> email_notify

Solution

1. Step 1: Define correct task order

   First, run models with dbt run, then test them with dbt test, and finally notify if tests pass.

2. Step 2: Set dependencies accordingly

   Use '>>' to chain tasks in order: dbt_run >> dbt_test >> email_notify.

3. Final Answer:

   dbt_run >> dbt_test >> email_notify -> Option D

4. Quick Check:

   Run then test then notify = D [OK]

Hint: Chain tasks in order with >>: run, test, then notify [OK]

Common Mistakes:

• Placing email notification before tests
• Reversing run and test order
• Skipping dependencies causing parallel runs