SqlOperator for database queries in Apache Airflow - Step-by-Step Execution

Process Flow - PostgresOperator for Postgres database queries

Define SQL query string

↓

Create PostgresOperator with query and connection

↓

Airflow schedules and triggers task

↓

PostgresOperator connects to DB

↓

Executes SQL query

↓

Commits transaction if necessary

↓

Task completes with success or failure

This flow shows how PostgresOperator runs a SQL query in Airflow: define query, create operator, Airflow runs it, connects to DB, executes query, and finishes.

Execution Sample

Apache Airflow

from airflow.providers.postgres.operators.postgres import PostgresOperator

sql_task = PostgresOperator(
    task_id='run_query',
    postgres_conn_id='my_postgres',
    sql='SELECT * FROM users;'
)

This code creates a PostgresOperator task that runs a SELECT query on a Postgres database using Airflow.

Process Table

Step	Action	Evaluation	Result
1	Define SQL query string	sql = 'SELECT * FROM users;'	Query string stored
2	Create PostgresOperator with task_id, postgres_conn_id, and sql	PostgresOperator(task_id='run_query', postgres_conn_id='my_postgres', sql=sql)	Operator instance created
3	Airflow scheduler triggers task	Task scheduled to run	Task starts execution
4	PostgresOperator connects to Postgres DB using 'my_postgres'	Connection established	DB connection ready
5	Execute SQL query 'SELECT * FROM users;'	Query sent to DB	DB executes query
6	Finalize query execution and transaction	Transaction handled	Execution complete
7	Task completes successfully	No errors	Task marked success

💡 Task finishes after successful query execution

Status Tracker

Variable	Start	After Step 1	After Step 2	After Step 4	After Step 6	Final
sql	undefined	SELECT * FROM users;	SELECT * FROM users;	SELECT * FROM users;	SELECT * FROM users;	SELECT * FROM users;
sql_task	undefined	undefined	PostgresOperator instance	PostgresOperator instance	PostgresOperator instance	PostgresOperator instance
db_connection	undefined	undefined	undefined	Connected	Connected	Closed after task
query_result	undefined	undefined	undefined	undefined	undefined	undefined

Key Moments - 3 Insights

Why do we need to specify 'postgres_conn_id' in PostgresOperator?

The 'postgres_conn_id' tells Airflow which database connection to use. Without it, PostgresOperator won't know where to run the query. See step 4 in execution_table where connection is established.

Does PostgresOperator automatically fetch query results?

No, PostgresOperator executes the query but does not automatically fetch or push results to XCom. It completes execution as shown in step 6. Use PythonOperator with hook.get_records() for SELECT data.

What happens if the SQL query has an error?

If the query fails, the task will fail and Airflow marks it as failed. This stops the flow unless retries or error handling are set. This is implied after step 6 if errors occur.

Visual Quiz - 3 Questions

Look at the execution table, at which step does PostgresOperator connect to the database?

A Step 4

B Step 2

C Step 6

D Step 1

💡 Hint

Check the 'Action' column for connection establishment in the execution_table.

According to variable_tracker, what is the value of 'query_result' after step 6?

A Fetched rows

B PostgresOperator instance

C undefined

D Closed after task

💡 Hint

Look at the 'query_result' row under 'After Step 6' in variable_tracker.

If the SQL query string changes, which step in execution_table reflects this change?

A Step 4

B Step 1

C Step 7

D Step 3

💡 Hint

Step 1 shows defining the SQL query string.

Concept Snapshot

PostgresOperator runs SQL queries in Airflow tasks.
Define SQL string, create operator with postgres_conn_id and SQL.
Airflow runs task, operator connects to DB, executes query.
Task completes success or failure.
Use postgres_conn_id to specify target database.

Full Transcript

PostgresOperator in Airflow lets you run SQL queries as tasks. First, you write your SQL query as a string. Then, you create a PostgresOperator instance, giving it a task ID, the postgres_conn_id, and the SQL query. When Airflow runs the task, PostgresOperator connects to the database using the connection ID. It sends the SQL query to the database and waits for completion. After the query runs, it handles the transaction and marks the task as successful if no errors occur. If the query fails, the task fails. This process allows you to automate database queries inside Airflow workflows easily.

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