SqlOperator for database queries in Apache Airflow - Mini Project: Build & Apply

Using SqlOperator in Airflow to Run Database Queries

📖 Scenario: You work as a data engineer and need to automate running SQL queries on a database using Airflow. Airflow helps schedule and run tasks automatically.In this project, you will create a simple Airflow DAG that uses SqlOperator to run a SQL query on a database.

🎯 Goal: Build an Airflow DAG that runs a SQL query using SqlOperator. You will define the DAG, set the SQL query, and print the query execution result.

📋 What You'll Learn

Create an Airflow DAG named example_sql_dag

Use SqlOperator to run a SQL query

Define a SQL query string variable sql_query with the exact query SELECT * FROM users WHERE active = 1;

Set the DAG schedule interval to @daily

Print a message after the SQL task runs

💡 Why This Matters

🌍 Real World

Automating database queries is common in data pipelines. Airflow helps schedule and run these queries regularly without manual work.

💼 Career

Data engineers and DevOps professionals use Airflow and SqlOperator to build reliable data workflows and automate database tasks.

1

Create the Airflow DAG and SQL query variable

Create an Airflow DAG named example_sql_dag with dag_id='example_sql_dag' and schedule_interval='@daily'. Also create a variable called sql_query and set it to the string "SELECT * FROM users WHERE active = 1;".

Apache Airflow

# Your code here

Need a hint?

Use DAG from airflow and set dag_id and schedule_interval. Define sql_query as the exact SQL string.

2

Import SqlOperator and create the SQL task

Import SqlOperator from airflow.providers.common.sql.operators.sql. Then create a task called run_sql using SqlOperator inside the example_sql_dag. Set task_id='run_sql', sql=sql_query, and dag=example_sql_dag.

Apache Airflow

from airflow import DAG
from datetime import datetime

example_sql_dag = DAG(dag_id='example_sql_dag', start_date=datetime(2024, 1, 1), schedule_interval='@daily')
sql_query = "SELECT * FROM users WHERE active = 1;"
# Your code here

Need a hint?

Import SqlOperator and create a task with the exact task_id, sql, and dag parameters.

3

Set task dependencies and add a print statement

Add a Python function called print_done that prints "SQL query task completed.". Then create a PythonOperator task called print_task with task_id='print_done' and python_callable=print_done inside example_sql_dag. Finally, set the task order so run_sql runs before print_task.

Apache Airflow

from airflow import DAG
from datetime import datetime
from airflow.providers.common.sql.operators.sql import SqlOperator
from airflow.operators.python import PythonOperator

example_sql_dag = DAG(dag_id='example_sql_dag', start_date=datetime(2024, 1, 1), schedule_interval='@daily')
sql_query = "SELECT * FROM users WHERE active = 1;"

run_sql = SqlOperator(task_id='run_sql', sql=sql_query, dag=example_sql_dag)
# Your code here

Need a hint?

Define a function that prints the message. Use PythonOperator to create a task with that function. Use >> to set task order.

4

Print the DAG tasks to verify setup

Add a print statement that prints the list of task IDs in example_sql_dag using list(example_sql_dag.task_ids).

Apache Airflow

from airflow import DAG
from datetime import datetime
from airflow.providers.common.sql.operators.sql import SqlOperator
from airflow.operators.python import PythonOperator

example_sql_dag = DAG(dag_id='example_sql_dag', start_date=datetime(2024, 1, 1), schedule_interval='@daily')
sql_query = "SELECT * FROM users WHERE active = 1;"

run_sql = SqlOperator(task_id='run_sql', sql=sql_query, dag=example_sql_dag)

def print_done():
    print("SQL query task completed.")

print_task = PythonOperator(task_id='print_done', python_callable=print_done, dag=example_sql_dag)

run_sql >> print_task
# Your code here

Need a hint?

Use print(list(example_sql_dag.task_ids)) to show the task IDs in the DAG.