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

Need a 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

Need a 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

Need a 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

Need a hint?

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