Orchestrating dbt with Airflow - Step-by-Step Execution

Concept Flow - Orchestrating dbt with Airflow

Start Airflow DAG

↓

Trigger dbt run task

↓

dbt runs models

↓

Check dbt run success?

No→Handle failure

Yes↓

Trigger dbt test task

↓

dbt runs tests

↓

Check dbt test success?

No→Handle failure

Yes↓

DAG completes successfully

Airflow starts a DAG that runs dbt models, checks success, then runs dbt tests, handling failures if any.

Execution Sample

dbt

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

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

dag = DAG('dbt_airflow', default_args=default_args, schedule_interval='@daily')

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

This Airflow DAG runs dbt models first, then runs dbt tests after the models finish.

Execution Table

Step	Task	Action	Result	Next Step
1	Start DAG	Trigger DAG run	DAG started	Run dbt_run task
2	dbt_run	Execute 'dbt run'	Models built successfully	Run dbt_test task
3	dbt_test	Execute 'dbt test'	Tests passed	DAG completes successfully
4	DAG Completion	All tasks done	Success	End

💡 DAG ends after dbt tests pass successfully

Variable Tracker

Variable	Start	After dbt_run	After dbt_test	Final
DAG Status	Not started	dbt_run success	dbt_test success	Success
dbt_run Status	Not started	Success	N/A	Success
dbt_test Status	Not started	N/A	Success	Success

Key Moments - 2 Insights

Why does dbt_test run only after dbt_run finishes?

Because in the execution_table, step 2 shows dbt_run completes before step 3 starts dbt_test, ensuring models exist before testing.

What happens if dbt_run fails?

The DAG stops and handles failure as shown by the 'Check dbt run success?' decision in the concept_flow, so dbt_test does not run.

Visual Quiz - 3 Questions

Look at the execution_table, what is the result of the dbt_run task at step 2?

A DAG started

B Tests passed

C Models built successfully

D Failure

💡 Hint

Check the 'Result' column for step 2 in the execution_table.

At which step does the DAG complete successfully?

A Step 2

B Step 4

C Step 3

D Step 1

💡 Hint

Look at the 'Task' and 'Result' columns in the execution_table for the final step.

If dbt_run fails, what happens to dbt_test?

A dbt_test is skipped

B DAG restarts from beginning

C dbt_test runs anyway

D dbt_test runs before dbt_run

💡 Hint

Refer to the concept_flow where failure after dbt_run leads to handling failure and no dbt_test run.

Concept Snapshot

Airflow DAG runs dbt tasks in order:
1. dbt run builds models
2. dbt test runs tests
Tasks run sequentially
Failures stop the DAG
Use BashOperator to run dbt commands

Full Transcript

This visual execution shows how Airflow orchestrates dbt workflows. The DAG starts and triggers the dbt run task to build models. After successful completion, it triggers the dbt test task to run tests on those models. If any task fails, the DAG stops and handles the failure. Variables track the status of each task and the overall DAG. This ensures dbt models are built before tests run, maintaining data pipeline integrity.

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