In the previous lesson, you created a simple DAG using the PythonOperator. Airflow, however, provides many built-in operators to handle common tasks such as running shell commands, executing SQL queries, or interacting with external systems.

Learning Objectives

By the end of this lesson, you will be able to:

  • Use PythonOperator to execute Python functions.

  • Use BashOperator to run shell commands.

  • Use SqlOperator to query a database.

  • Combine operators to create multi-step workflows.

1. The PythonOperator

  • Runs a Python function directly inside an Airflow task.

  • Best for logic implemented in Python scripts.

Example:

from airflow import DAG
from airflow.operators.python import PythonOperator
from datetime import datetime

def greet(name):
    print(f"Hello, {name}!")

with DAG(
    dag_id="python_operator_example",
    start_date=datetime(2025, 1, 1),
    schedule_interval="@daily",
    catchup=False
) as dag:

    task1 = PythonOperator(
        task_id="greet_task",
        python_callable=greet,
        op_args=["Airflow Learner"]   # pass arguments
    )

2. The BashOperator

  • Executes shell commands or scripts.

  • Useful for system tasks, file operations, or invoking CLI tools.

Example:

from airflow.operators.bash import BashOperator

task2 = BashOperator(
    task_id="bash_task",
    bash_command="echo 'Today is $(date)'"
)

When run, this prints the current date in the logs.

3. The SqlOperator

  • Executes SQL queries on a connected database.

  • Requires a database connection set up in Airflow Connections.

Example (using Postgres):

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

task3 = PostgresOperator(
    task_id="sql_task",
    postgres_conn_id="my_postgres",
    sql="CREATE TABLE IF NOT EXISTS users (id SERIAL PRIMARY KEY, name TEXT);"
)

Here, Airflow connects to Postgres via my_postgres (configured in the UI) and runs the SQL statement.

4. Combining Operators in a DAG

with DAG(
    dag_id="multi_operator_example",
    start_date=datetime(2025, 1, 1),
    schedule_interval="@daily",
    catchup=False
) as dag:

    # Python task
    greet_task = PythonOperator(
        task_id="greet_task",
        python_callable=greet,
        op_args=["Airflow"]
    )

    # Bash task
    date_task = BashOperator(
        task_id="date_task",
        bash_command="date"
    )

    # SQL task
    create_table_task = PostgresOperator(
        task_id="create_table_task",
        postgres_conn_id="my_postgres",
        sql="CREATE TABLE IF NOT EXISTS logs (id SERIAL PRIMARY KEY, ts TIMESTAMP);"
    )

    # Define dependencies
    greet_task >> date_task >> create_table_task

👉 Workflow order: Greet → Print Date → Create Table.

5. Best Practices

  • Use PythonOperator for Python logic.

  • Use BashOperator for lightweight shell tasks.

  • Use SqlOperator for database operations.

  • Don’t overload Bash or Python tasks with heavy processing – use external systems like Spark or DB engines.

  • Keep DAGs readable by grouping related tasks logically.

Lesson Summary

  • PythonOperator → runs Python functions.

  • BashOperator → runs shell commands.

  • SqlOperator → executes SQL queries against databases.

  • Operators are the building blocks of Airflow workflows.

  • Combining operators allows you to orchestrate multi-step, cross-system workflows.

PreviousNext