Airflow Operators Guide

An Operator defines what a task actually does. When you add a task to a DAG, you choose an operator that matches the job — run a Python function, execute a SQL query, send an email, or call an API. Airflow ships with dozens of built-in operators, and thousands more come from community providers.

What Is an Operator?

Think of an operator as a job template. You pick the template that fits your work, fill in the details, and Airflow handles the execution.

Real World Analogy:
──────────────────
Operator = Job Role
Task     = Specific Person Doing That Job

"Email Operator" is the role.
"Send sales report to manager@company.com" is the specific task using that role.

The Three Most Common Operators

1. PythonOperator

Runs any Python function you write. It is the most flexible operator.

from airflow.operators.python import PythonOperator

def calculate_total():
    total = 100 + 200
    print(f"Total: {total}")

calculate_task = PythonOperator(
    task_id="calculate_total",
    python_callable=calculate_total,
)

Use PythonOperator when your task is custom logic — data transformation, API calls, file processing, or any Python code.

2. BashOperator

Runs a shell command or shell script.

from airflow.operators.bash import BashOperator

backup_task = BashOperator(
    task_id="backup_files",
    bash_command="cp /data/sales.csv /backup/sales_$(date +%Y%m%d).csv",
)

Use BashOperator when you need to run command-line tools, shell scripts, or system commands.

3. EmailOperator

Sends an email from within a DAG.

from airflow.operators.email import EmailOperator

notify_task = EmailOperator(
    task_id="send_report_email",
    to="manager@company.com",
    subject="Daily Sales Report Ready",
    html_content="<p>The daily sales report has been processed.</p>",
)

Use EmailOperator to notify people when a pipeline finishes, fails, or produces a result.

Database Operators

PostgresOperator

Executes SQL queries on a PostgreSQL database.

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

create_table = PostgresOperator(
    task_id="create_sales_table",
    postgres_conn_id="my_postgres_db",
    sql="""
        CREATE TABLE IF NOT EXISTS daily_sales (
            id SERIAL PRIMARY KEY,
            sale_date DATE,
            amount NUMERIC
        );
    """,
)

The postgres_conn_id references a connection you define in Admin → Connections in the Airflow UI. Your password never appears in the code.

BigQueryInsertJobOperator

Runs SQL queries on Google BigQuery. Similar pattern — you pass a SQL string and a connection ID.

Cloud Storage Operators

S3ToGCSOperator

Copies a file from Amazon S3 to Google Cloud Storage in one task — no custom code needed.

from airflow.providers.google.cloud.transfers.s3_to_gcs import S3ToGCSOperator

move_file = S3ToGCSOperator(
    task_id="copy_to_gcs",
    bucket="my-s3-bucket",
    prefix="reports/",
    dest_gcs="gs://my-gcs-bucket/reports/",
    aws_conn_id="aws_default",
    gcp_conn_id="google_cloud_default",
)

Sensor Operators (Wait and Watch)

A Sensor is a special type of operator that waits for a condition to be true before the DAG moves on. Sensors check the condition repeatedly until it passes or times out.

DAG Flow With a Sensor:
─────────────────────────────────────────────
[Sensor: Wait for file to appear in S3]
    │  (checks every 60 seconds for up to 1 hour)
    │  FILE FOUND ✓
    ▼
[Download File]
    ▼
[Process File]
─────────────────────────────────────────────

FileSensor

from airflow.sensors.filesystem import FileSensor

wait_for_file = FileSensor(
    task_id="wait_for_report_csv",
    filepath="/data/reports/daily.csv",
    poke_interval=60,       # check every 60 seconds
    timeout=3600,           # give up after 1 hour
    mode="poke",
)

HttpSensor

Waits until an HTTP endpoint returns a success response.

from airflow.providers.http.sensors.http import HttpSensor

wait_for_api = HttpSensor(
    task_id="wait_for_api_ready",
    http_conn_id="my_api",
    endpoint="/health",
    poke_interval=30,
)

DummyOperator (EmptyOperator)

Does nothing. Use it as a marker task to organize complex DAG structures:

from airflow.operators.empty import EmptyOperator

start = EmptyOperator(task_id="pipeline_start")
end   = EmptyOperator(task_id="pipeline_end")

start >> [task_a, task_b, task_c] >> end

                [start]
               /   |   \
        [task_a] [task_b] [task_c]
               \   |   /
                 [end]

The start and end markers make the DAG easier to read in the UI without doing any actual work.

Choosing the Right Operator

Installing Provider Packages

Database and cloud operators come in separate installable packages called providers. Install only the providers you need:

# For PostgreSQL
pip install apache-airflow-providers-postgres

# For Google Cloud
pip install apache-airflow-providers-google

# For Amazon AWS
pip install apache-airflow-providers-amazon

After installing, restart the Airflow webserver and scheduler so Airflow detects the new operators.