Integration with dbt and Airflow in Snowflake - Mini Project: Build & Apply

Integration with dbt and Airflow

📖 Scenario: You work in a data engineering team. Your team uses dbt to transform data in Snowflake and Airflow to schedule and manage workflows. You need to set up a simple Airflow DAG that triggers a dbt job to run a model in Snowflake.

🎯 Goal: Build an Airflow DAG that runs a dbt command to execute a model in Snowflake. The DAG should include a task to run the dbt model and a task to check the success of the run.

📋 What You'll Learn

Create a Snowflake connection dictionary with required credentials.

Define a dbt command string to run a specific model.

Create an Airflow DAG with two tasks: one to run the dbt command and one to check success.

Use Airflow's BashOperator to run the dbt command.

💡 Why This Matters

🌍 Real World

Data teams use dbt to transform data in cloud warehouses like Snowflake. Airflow schedules and manages these transformation jobs to run automatically and reliably.

💼 Career

Understanding how to integrate dbt with Airflow and Snowflake is essential for data engineers and analysts to build automated, maintainable data pipelines.

1

Create Snowflake connection dictionary

Create a dictionary called snowflake_conn with these exact keys and values: 'account': 'xy12345', 'user': 'data_eng', 'password': 'SafePass123', 'warehouse': 'COMPUTE_WH', 'database': 'ANALYTICS_DB', 'schema': 'PUBLIC'.

Snowflake

# Create the Snowflake connection dictionary
# Your code here

Hint

Use curly braces to create a dictionary with the exact keys and values.

2

Define dbt run command string

Create a string variable called dbt_run_command that contains the exact command "dbt run --models my_model --profiles-dir ./profiles".

Snowflake

snowflake_conn = {
    'account': 'xy12345',
    'user': 'data_eng',
    'password': 'SafePass123',
    'warehouse': 'COMPUTE_WH',
    'database': 'ANALYTICS_DB',
    'schema': 'PUBLIC'
}

# Define the dbt run command string
# Your code here

Hint

Assign the exact dbt command string to the variable dbt_run_command.

3

Create Airflow DAG with BashOperator to run dbt

Import DAG from airflow and BashOperator from airflow.operators.bash. Create a DAG called dbt_snowflake_dag with dag_id='dbt_snowflake_dag', start_date=datetime(2024, 1, 1), and schedule_interval='@daily'. Inside the DAG context, create a task called run_dbt using BashOperator with task_id='run_dbt' and bash_command=dbt_run_command.

Snowflake

snowflake_conn = {
    'account': 'xy12345',
    'user': 'data_eng',
    'password': 'SafePass123',
    'warehouse': 'COMPUTE_WH',
    'database': 'ANALYTICS_DB',
    'schema': 'PUBLIC'
}

dbt_run_command = "dbt run --models my_model --profiles-dir ./profiles"

# Import DAG and BashOperator
# Create DAG and run_dbt task
# Your code here

Hint

Use the with DAG(...) syntax and create a BashOperator task inside it.

4

Add success check task and set task dependencies

Inside the existing DAG context, create a BashOperator task called check_success with task_id='check_success' and bash_command='echo Success'. Set the task dependency so that run_dbt runs before check_success.

Snowflake

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

snowflake_conn = {
    'account': 'xy12345',
    'user': 'data_eng',
    'password': 'SafePass123',
    'warehouse': 'COMPUTE_WH',
    'database': 'ANALYTICS_DB',
    'schema': 'PUBLIC'
}

dbt_run_command = "dbt run --models my_model --profiles-dir ./profiles"

with DAG(dag_id='dbt_snowflake_dag', start_date=datetime(2024, 1, 1), schedule_interval='@daily') as dag:
    run_dbt = BashOperator(
        task_id='run_dbt',
        bash_command=dbt_run_command
    )

    # Create check_success task and set dependencies
    # Your code here

Hint

Create the second task and use the >> operator to set the order.

Practice

1. What is the main role of dbt when integrated with Snowflake?

easy

A. Transforming raw data into clean, modeled tables using SQL

B. Scheduling data pipeline tasks automatically

C. Storing data securely in Snowflake

D. Monitoring Snowflake warehouse usage

Solution

1. Step 1: Understand dbt's purpose

   dbt is designed to transform raw data into clean, modeled tables using SQL scripts.

2. Step 2: Differentiate from Airflow's role

   Airflow handles scheduling and automation, not data transformation itself.

3. Final Answer:

   Transforming raw data into clean, modeled tables using SQL -> Option A

4. Quick Check:

   dbt = data transformation [OK]

Hint: Remember: dbt transforms data, Airflow schedules tasks [OK]

Common Mistakes:

• Confusing Airflow's scheduling role with dbt's transformation role
• Thinking dbt stores data instead of transforming it
• Assuming dbt monitors Snowflake usage

2. Which file in dbt contains the Snowflake connection details for integration?

easy

A. airflow.cfg

B. dbt_project.yml

C. profiles.yml

D. snowflake.sql

Solution

1. Step 1: Identify dbt configuration files

   dbt uses profiles.yml to store connection details like Snowflake credentials.

2. Step 2: Exclude unrelated files

   dbt_project.yml configures project settings, airflow.cfg is for Airflow, and snowflake.sql is not a config file.

3. Final Answer:

   profiles.yml -> Option C

4. Quick Check:

   Connection details = profiles.yml [OK]

Hint: Connection info in dbt is always in profiles.yml [OK]

Common Mistakes:

• Confusing dbt_project.yml with profiles.yml
• Mixing Airflow config with dbt config
• Assuming SQL files hold connection info

3. Given this Airflow DAG snippet to run a dbt command:

dbt_run = BashOperator(
    task_id='dbt_run',
    bash_command='dbt run --profiles-dir /home/airflow/.dbt',
    dag=dag
)

What happens when this task runs?

medium

A. Airflow executes dbt models transforming data in Snowflake

B. Airflow creates a new Snowflake warehouse

C. dbt schedules the Airflow DAG

D. Airflow deletes old Snowflake tables

Solution

1. Step 1: Analyze the BashOperator command

   The command dbt run triggers dbt to execute SQL models and transform data in Snowflake.

2. Step 2: Understand Airflow's role

   Airflow runs this command as a task; it does not create warehouses or delete tables itself.

3. Final Answer:

   Airflow executes dbt models transforming data in Snowflake -> Option A

4. Quick Check:

   dbt run = data transformation task [OK]

Hint: dbt run transforms data; Airflow just runs commands [OK]

Common Mistakes:

• Thinking Airflow creates or deletes Snowflake resources
• Confusing dbt scheduling with Airflow scheduling
• Assuming dbt controls Airflow DAGs

4. You configured Airflow to run dbt commands but the tasks fail with authentication errors. What is the most likely cause?

medium

A. Airflow DAG file missing task_id

B. Snowflake warehouse is paused

C. dbt models have syntax errors

D. Incorrect Snowflake credentials in dbt's profiles.yml

Solution

1. Step 1: Identify cause of authentication errors

   Authentication errors usually mean wrong or missing credentials in profiles.yml used by dbt.

2. Step 2: Exclude other causes

   Missing task_id causes Airflow errors but not authentication errors; syntax errors cause runtime failures; paused warehouse causes query delays, not auth errors.

3. Final Answer:

   Incorrect Snowflake credentials in dbt's profiles.yml -> Option D

4. Quick Check:

   Auth error = wrong credentials [OK]

Hint: Check profiles.yml credentials first on auth errors [OK]

Common Mistakes:

• Blaming Airflow config instead of dbt credentials
• Assuming syntax errors cause authentication failures
• Ignoring warehouse state impact

5. You want to automate a data pipeline that runs dbt transformations daily and sends email alerts on failure using Airflow. Which setup best achieves this?

hard

A. Schedule dbt runs in Snowflake and use Airflow only for monitoring

B. Create an Airflow DAG with dbt BashOperator tasks and EmailOperator for alerts

C. Use only dbt Cloud scheduler with no Airflow integration

D. Manually run dbt commands and check logs for errors

Solution

1. Step 1: Identify automation components

   Airflow DAG can schedule dbt commands using BashOperator and send emails on failure with EmailOperator.

2. Step 2: Evaluate other options

   Scheduling dbt in Snowflake alone lacks alerting; dbt Cloud scheduler excludes Airflow's flexibility; manual runs lack automation.

3. Final Answer:

   Create an Airflow DAG with dbt BashOperator tasks and EmailOperator for alerts -> Option B

4. Quick Check:

   Airflow DAG + dbt + email = automation [OK]

Hint: Use Airflow DAG with dbt tasks and email alerts for automation [OK]

Common Mistakes:

• Ignoring Airflow's alerting capabilities
• Relying solely on manual or Snowflake scheduling
• Confusing dbt Cloud scheduler with Airflow