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

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

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

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

Hint

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

Practice

1. What is the primary purpose of the SqlOperator in Airflow?

easy

A. To run SQL commands as part of an Airflow task

B. To schedule Python scripts

C. To monitor server health

D. To send email notifications

Solution

1. Step 1: Understand the role of SqlOperator

   The SqlOperator is designed to execute SQL commands within Airflow tasks, automating database operations.

2. Step 2: Compare with other options

   Scheduling Python scripts, monitoring servers, and sending emails are handled by other operators or components, not SqlOperator.

3. Final Answer:

   To run SQL commands as part of an Airflow task -> Option A

4. Quick Check:

   SqlOperator runs SQL commands = D [OK]

Hint: SqlOperator always runs SQL queries inside tasks [OK]

Common Mistakes:

• Confusing SqlOperator with PythonOperator
• Thinking SqlOperator schedules scripts
• Assuming SqlOperator sends emails

2. Which of the following is the correct way to create a SqlOperator task in Airflow?

easy

A. SqlOperator(id='run_query', query='SELECT * FROM table', connection='my_db')

B. SqlOperator(name='run_query', sql_command='SELECT * FROM table', conn='my_db')

C. SqlOperator(task='run_query', command='SELECT * FROM table', connection_id='my_db')

D. SqlOperator(task_id='run_query', sql='SELECT * FROM table', conn_id='my_db')

Solution

1. Step 1: Check parameter names for SqlOperator

   The correct parameters are task_id, sql, and conn_id.

2. Step 2: Validate each option

   Only SqlOperator(task_id='run_query', sql='SELECT * FROM table', conn_id='my_db') uses the correct parameter names and syntax for SqlOperator.

3. Final Answer:

   SqlOperator(task_id='run_query', sql='SELECT * FROM table', conn_id='my_db') -> Option D

4. Quick Check:

   Correct parameters = B [OK]

Hint: Remember: task_id, sql, conn_id are required parameters [OK]

Common Mistakes:

• Using wrong parameter names like 'id' or 'query'
• Mixing connection parameter names
• Omitting task_id or sql

3. Given this Airflow task using SqlOperator:

task = SqlOperator(
    task_id='count_users',
    sql='SELECT COUNT(*) FROM users;',
    conn_id='my_postgres'
)

What will be the main effect when this task runs?

medium

A. It will create a new table named 'count_users'

B. It will return the count of users and store it in Airflow logs

C. It will delete all records from the users table

D. It will send an email with the user count

Solution

1. Step 1: Understand the SQL command

   The SQL command SELECT COUNT(*) FROM users; counts the number of rows in the users table.

2. Step 2: Understand SqlOperator behavior

   SqlOperator runs the query and logs the output; it does not create tables, delete data, or send emails by default.

3. Final Answer:

   It will return the count of users and store it in Airflow logs -> Option B

4. Quick Check:

   SELECT COUNT(*) logs result = A [OK]

Hint: SELECT queries log results; no table creation or deletion [OK]

Common Mistakes:

• Thinking SELECT creates or deletes tables
• Assuming SqlOperator sends emails automatically
• Confusing query effects with task side effects

4. You wrote this SqlOperator task:

task = SqlOperator(
    task_id='update_status',
    sql='UPDATE orders SET status = 'shipped' WHERE id = 10;',
    conn_id='my_db'
)

But Airflow throws a syntax error. What is the problem?

medium

A. conn_id should be connection_id

B. The task_id is missing

C. The SQL string uses single quotes inside single quotes without escaping

D. SqlOperator cannot run UPDATE statements

Solution

1. Step 1: Check the SQL string syntax

   The SQL string uses single quotes around the whole string and also inside for 'shipped', causing a conflict.

2. Step 2: Understand string escaping in Python

   Inner single quotes must be escaped or the string should use double quotes outside to avoid syntax errors.

3. Final Answer:

   The SQL string uses single quotes inside single quotes without escaping -> Option C

4. Quick Check:

   Escape inner quotes or use double quotes outside [OK]

Hint: Escape inner quotes or use double quotes outside SQL string [OK]

Common Mistakes:

• Ignoring string quoting rules in Python
• Changing parameter names incorrectly
• Thinking SqlOperator can't run UPDATE

5. You want to run multiple SQL commands in one SqlOperator task to reset a table:

DROP TABLE IF EXISTS temp_data;
CREATE TABLE temp_data (id INT, value TEXT);
INSERT INTO temp_data VALUES (1, 'A'), (2, 'B');

How should you write the sql parameter to run all commands correctly?

hard

A. Use a single string with commands separated by semicolons

B. Pass a list of strings, one per command

C. Use multiple SqlOperator tasks, one per command

D. Put all commands in a file and pass the filename to sql

Solution

1. Step 1: Understand SqlOperator's sql parameter

   SqlOperator accepts a single string with multiple SQL commands separated by semicolons.

2. Step 2: Compare other options

   Passing a list is invalid; multiple tasks or a file are alternatives but not the direct answer to the question.

3. Final Answer:

   Use a single string with commands separated by semicolons -> Option A

4. Quick Check:

   Multiple SQL commands in one string separated by semicolons [OK]

Hint: Separate SQL commands with semicolons in one string [OK]

Common Mistakes:

• Passing list instead of string to sql
• Trying to run multiple commands without semicolons
• Confusing file input with direct sql parameter