SqlOperator for database queries in Apache Airflow - Practice Problems & Coding Challenges

Challenge - 5 Problems

💻 Command Output

intermediate

What is the output of this Airflow SqlOperator task?

Given the following Airflow DAG snippet, what will be the output in the Airflow logs when the task runs successfully?

Apache Airflow

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

default_args = {'start_date': datetime(2024, 1, 1)}

dag = DAG('test_sql_operator', default_args=default_args, schedule_interval='@once')

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

A The task will succeed but no output will be shown in logs by default.

B The task will fail because SqlOperator cannot run SELECT queries.

C The task logs will show the count of rows in the users table as a result.

D The task will raise a syntax error due to missing parameters.

💡 Hint

Think about what SqlOperator does with SELECT queries and how results are handled.

The SqlOperator runs the SQL query but does not fetch or log the results by default. It executes the query on the database and marks the task as successful if no error occurs. To see query results, you need to fetch them explicitly or use a different operator.

❓ Configuration

intermediate

Which configuration correctly runs a SQL script file using SqlOperator?

You want to run a SQL script stored in '/sql_scripts/create_tables.sql' using Airflow's SqlOperator. Which option correctly configures the operator?

A PostgresOperator(task_id='create_tables', postgres_conn_id='my_postgres', sql=open('/sql_scripts/create_tables.sql').read(), dag=dag)

B PostgresOperator(task_id='create_tables', postgres_conn_id='my_postgres', sql='RUN /sql_scripts/create_tables.sql', dag=dag)

C PostgresOperator(task_id='create_tables', postgres_conn_id='my_postgres', sql='@/sql_scripts/create_tables.sql', dag=dag)

D PostgresOperator(task_id='create_tables', postgres_conn_id='my_postgres', sql='/sql_scripts/create_tables.sql', dag=dag)

💡 Hint

SqlOperator's sql parameter accepts a string containing the SQL commands.

The sql parameter expects the SQL commands as a string. Reading the file content and passing it as a string is the correct way. Passing the file path as a string or using SQL commands like RUN will not work.

❓ Troubleshoot

advanced

Why does this SqlOperator task fail with 'could not connect to server' error?

You have this task: PostgresOperator(task_id='check_data', postgres_conn_id='wrong_conn', sql='SELECT 1;', dag=dag) It fails with 'could not connect to server' error. What is the most likely cause?

A The DAG start_date is in the future, so the task cannot connect.

B The SQL query 'SELECT 1;' is invalid and causes connection failure.

C The postgres_conn_id 'wrong_conn' is not defined or has wrong connection details in Airflow connections.

D The PostgresOperator requires extra parameters to connect, missing here.

💡 Hint

Check the connection ID and its configuration in Airflow UI.

The error 'could not connect to server' usually means the connection details are wrong or missing. The connection ID must be correctly set up in Airflow connections with valid host, port, user, and password.

🔀 Workflow

advanced

How to chain multiple SqlOperator tasks to run sequentially?

You have three SQL tasks: create_table, insert_data, and verify_data. How do you set them to run one after another in Airflow?

Apache Airflow

create_table = PostgresOperator(task_id='create_table', postgres_conn_id='my_postgres', sql='CREATE TABLE test(id INT);', dag=dag)
insert_data = PostgresOperator(task_id='insert_data', postgres_conn_id='my_postgres', sql='INSERT INTO test VALUES (1);', dag=dag)
verify_data = PostgresOperator(task_id='verify_data', postgres_conn_id='my_postgres', sql='SELECT COUNT(*) FROM test;', dag=dag)

A create_table | insert_data | verify_data

B create_table & insert_data & verify_data

C create_table + insert_data + verify_data

D create_table >> insert_data >> verify_data

💡 Hint

Airflow uses bitshift operators to set task dependencies.

In Airflow, the '>>' operator sets downstream dependencies, so tasks run sequentially in that order.

✅ Best Practice

expert

What is the best practice to handle sensitive database credentials with SqlOperator?

You want to run SQL queries securely in Airflow without exposing database passwords in DAG code. What is the best practice?

A Save credentials in a plain text file and read them in the DAG code.

B Store credentials in Airflow Connections and reference them by connection ID in SqlOperator.

C Hardcode the username and password directly in the sql parameter string.

D Use environment variables inside the sql parameter string to pass credentials.

💡 Hint

Airflow has a built-in way to manage secrets and connections.

Airflow Connections securely store credentials and allow operators to reference them by ID. This avoids exposing secrets in code or files.