6. Operators, sensors and hooks

A task is an instance of an operator. Everything Airflow can do is an operator someone wrote, and there are hundreds — so the useful knowledge is which handful you actually need and which patterns to avoid.

The three kinds of thing

   OPERATOR   does something and finishes
              BashOperator, PythonOperator,
              KubernetesPodOperator, SQLExecuteQueryOperator

   SENSOR     waits for something to become true,
              then finishes
              FileSensor, ExternalTaskSensor, S3KeySensor

   HOOK       the connection layer operators are built on
              PostgresHook, S3Hook, HttpHook

A hook is what you use inside a @task when no operator fits. It handles authentication from an Airflow Connection and gives you a client:

@task
def load():
    from airflow.providers.postgres.hooks.postgres import PostgresHook
    hook = PostgresHook(postgres_conn_id="warehouse")
    hook.run("INSERT INTO summary SELECT ... ")

Hooks are underused. People write psycopg2.connect(os.environ["DB_URL"]) inside tasks and lose connection management, credential handling and the ability to change a connection without redeploying.

Providers

Since 2.0, everything except a small core lives in separately versioned provider packages:

pip install apache-airflow-providers-amazon
pip install apache-airflow-providers-google
pip install apache-airflow-providers-postgres
pip install apache-airflow-providers-cncf-kubernetes
pip install apache-airflow-providers-slack

This is good — providers release independently of Airflow — and it produces the classic first error:

   ModuleNotFoundError: No module named
     'airflow.providers.amazon'

which means the provider is not installed, not that your code is wrong.

Pin provider versions in production. They release frequently and occasionally change operator arguments in ways that are minor for them and breaking for you.

The operators worth knowing

BashOperator — run a shell command. Templated, so {{ ds }} works. The simplest thing that could possibly work, and often correct.

PythonOperator / @task — run a Python function. Use @task.

SQLExecuteQueryOperator — the modern unified SQL operator, replacing the per-database ones (PostgresOperator, MySqlOperator, SnowflakeOperator, all deprecated).

from airflow.providers.common.sql.operators.sql import SQLExecuteQueryOperator

SQLExecuteQueryOperator(
    task_id="aggregate",
    conn_id="warehouse",
    sql="sql/daily_agg.sql",       # a FILE — templated, reviewable
    params={"table": "orders"},
)

Putting SQL in .sql files and pointing at them is much better than embedding it in Python strings: syntax highlighting, review diffs that make sense, and reuse.

KubernetesPodOperator — run a container as a task. Discussed below; the most important operator in the list.

TriggerDagRunOperator — start another DAG. Prefer assets (chapter 4) when the relationship is about data.

EmptyOperator — does nothing. Useful as a join point or a named milestone in the graph.

@task.branch — choose which downstream path to follow.

@task.short_circuit — return False and everything downstream is skipped. The clean way to say “there is no data today, stop”.

@task.bash, @task.docker, @task.kubernetes, @task.virtualenv — decorator forms, generally nicer than the operator classes.

KubernetesPodOperator and the dependency problem

This deserves its own section because it solves Airflow’s worst structural problem.

Every Python dependency any DAG needs must be installed in the Airflow image. One DAG needs pandas 1.5; another needs pandas 2.2. One needs an old boto3; another needs a new one. There is one image, so there is one answer, and teams end up negotiating dependency versions across unrelated pipelines. This is genuinely miserable at scale.

from airflow.providers.cncf.kubernetes.operators.pod import KubernetesPodOperator

KubernetesPodOperator(
    task_id="train_model",
    name="train-model",
    image="acme/ml-training:v2.1.0",      # ← its OWN dependencies
    cmds=["python", "train.py"],
    arguments=["--date", "{{ ds }}"],
    container_resources=k8s.V1ResourceRequirements(
        requests={"memory": "8Gi", "cpu": "2"},
        limits={"memory": "16Gi"},
    ),
    get_logs=True,
    is_delete_operator_pod=True,
)

What this buys you:

Dependency isolation. Each task’s container has exactly what it needs. Airflow’s image stays small and stable.

Resource isolation. A memory-hungry task gets a pod with a limit; it cannot take a worker down with it.

Language freedom. The task is a container. It does not have to be Python.

Airflow stops being the compute. Which is chapter 1’s rule, enforced structurally rather than by discipline.

The cost is pod startup: 1–30 seconds depending on image size and whether it is cached on the node. For a five-minute task that is noise; for a five-second task it is absurd. Use it for substantial work.

If you run Airflow on Kubernetes, this is usually the right default for anything non-trivial. Airflow becomes a scheduler that starts containers, which is a much healthier architecture than Airflow-as-runtime.

Sensors, and the mode that eats your workers

A sensor waits. The mode argument decides how, and the default is the expensive one.

FileSensor(
    task_id="wait_for_file",
    filepath="/data/incoming/{{ ds }}.csv",
    mode="poke",            # ← DEFAULT
    poke_interval=60,
    timeout=60 * 60 * 6,
)

Mode	Behaviour	Worker slot
poke	loop and sleep inside the task	held the entire time
reschedule	check, exit, be rescheduled later	released between checks
deferrable	hand to the triggerer	released entirely

   20 sensors × 6 hours in poke mode
     = 20 worker slots doing nothing for 6 hours

   the same in reschedule mode
     = 20 slots for a few seconds every minute

   the same as deferrable operators
     = 0 slots; one triggerer handles all of them

This is the single most common cause of “all my workers are busy and nothing is running.” A pool of 32 workers with 30 sensors in poke mode has two usable slots, and the symptom — tasks stuck in queued — looks like a capacity problem.

The rules:

Always set a timeout. Without one a sensor waits forever, and forever is a long time to hold a worker slot.

Use mode="reschedule" for anything waiting more than a couple of minutes, unless poke_interval is very short.

Prefer a deferrable version where one exists — S3KeySensorAsync, deferrable=True on many operators. Chapter 11.

Question whether you need a sensor at all. Sensors are polling, and polling is what you do when nothing will tell you. If the producer is another Airflow DAG, an asset (chapter 4) is strictly better: event-driven, no polling, no timeout to tune.

ExternalTaskSensor, specifically

Waiting for a task in another DAG. It works and it is fragile, because it matches on exact logical dates:

ExternalTaskSensor(
    task_id="wait_upstream",
    external_dag_id="ingest",
    external_task_id="finish",
    execution_delta=timedelta(hours=1),   # ← if schedules differ
    mode="reschedule",
    timeout=3600,
)

If the two DAGs have different schedules you must compute execution_delta or execution_date_fn correctly, and getting it wrong produces a sensor that waits forever for a run that will never exist by that name. Assets replaced this pattern and are better in every respect. Reach for ExternalTaskSensor only when you cannot change the producer.

Writing a custom operator

Usually you should not — a @task with a hook does the same job with less ceremony. Write one when the same logic appears in many DAGs:

from airflow.models import BaseOperator

class WarehouseLoadOperator(BaseOperator):
    template_fields = ("table", "date")          # ← enables Jinja

    def __init__(self, *, table: str, date: str, conn_id: str, **kwargs):
        super().__init__(**kwargs)
        self.table, self.date, self.conn_id = table, date, conn_id

    def execute(self, context):
        hook = PostgresHook(postgres_conn_id=self.conn_id)
        hook.run(f"CALL load_{self.table}(%s)", parameters=(self.date,))
        return {"table": self.table}            # goes to XCom

Two things that are easy to miss:

template_fields is what makes Jinja work. A field not listed there receives the literal string {{ ds }}, which produces a wonderfully confusing bug.

Do the work in execute, not __init__. __init__ runs at DAG parse time — every thirty seconds, forever, on every worker. Chapter 3’s rule applies inside operators too, and this is the version of it people miss.

What to take from this chapter

• Operators do; sensors wait; hooks are the connection layer. Use a hook inside @task when no operator fits — hooks are underused.
• Providers are separate packages. ModuleNotFoundError: airflow.providers.x means it is not installed. Pin provider versions.
• Use SQLExecuteQueryOperator with SQL in .sql files, not the deprecated per-database operators.
• KubernetesPodOperator solves Airflow’s dependency problem: each task brings its own image, its own libraries and its own resource limits. Costs 1–30 s of pod startup. On Kubernetes it is the right default for substantial tasks.
• Sensors in the default poke mode hold a worker slot the entire time. This is the most common cause of “all workers busy, nothing running”. Use reschedule, or better, deferrable.
• Always set a sensor timeout.
• ExternalTaskSensor matches on exact logical dates and is fragile. Assets replaced it and are better.
• In a custom operator, template_fields enables Jinja, and __init__ runs at parse time — do the work in execute.

Next: the executors — Local, Celery, Kubernetes and Edge, and how to choose between them.