Operators and Tasks

An operator is a reusable template for a unit of work. A task is an operator with its parameters supplied, placed in a DAG. PythonOperator with task_id="load_staging" and a python_callable is one task; the same operator class with different parameters is a different task.

PythonOperator#

The workhorse for data pipelines. It runs a Python function as a task.

from airflow.operators.python import PythonOperator
from transform.raw_to_staging import load_raw_orders_to_staging
load_staging = PythonOperator(    task_id="load_raw_orders_to_staging",    python_callable=load_raw_orders_to_staging,)

Passing arguments#

If the function needs parameters, supply them with op_args (positional) or op_kwargs (keyword):

def load_table(source_db, target_table, batch_size=1000):    ...
load_orders = PythonOperator(    task_id="load_orders",    python_callable=load_table,    op_kwargs={        "source_db": "raw",        "target_table": "staging.orders",        "batch_size": 5000,    },)

Keep these to small configuration values. They are rendered and stored by Airflow, so they are not the place for large data.

The task context#

Airflow can pass runtime information into the callable. In current Airflow the function just declares **context (or the specific keys it wants):

def load_incrementally(**context):    data_interval_start = context["data_interval_start"]    data_interval_end = context["data_interval_end"]
    # load only rows whose timestamp falls in [start, end)    load_window(start=data_interval_start, end=data_interval_end)

load_window_task = PythonOperator(    task_id="load_window",    python_callable=load_incrementally,)

Useful context keys:

Use the data interval, not datetime.now(). A run for the 09:00–10:00 interval should process 09:00–10:00 data whenever it actually executes — on time, an hour late, or during a backfill. Reading now() makes the result depend on wall-clock time and breaks reruns.

The @task decorator (TaskFlow)#

Modern Airflow offers a shorter form. Decorating a function with @task turns it into a task; calling it in the DAG wires it up, and its return value is passed to whatever consumes it:

from airflow.decorators import task
@taskdef load_raw_orders_to_staging():    ...
@taskdef build_customer_metrics():    ...
with DAG(...) as dag:    load_raw_orders_to_staging() >> build_customer_metrics()

TaskFlow is the recommended style for new Python-heavy DAGs — it removes the PythonOperator boilerplate and makes passing values between tasks natural. PythonOperator is still everywhere in existing code, so both are worth knowing. The concepts below apply to either.

TaskFlow, a little deeper#

With TaskFlow, a task's return value flows to whichever task uses it, and Airflow inserts the XCom push/pull for you:

from airflow.decorators import dag, taskfrom datetime import datetime
@dag(schedule="@hourly", start_date=datetime(2026, 1, 1), catchup=False)def orders_pipeline():
    @task    def read_watermark() -> str:        ...        return watermark
    @task    def load_since(watermark: str):        ...
    load_since(read_watermark())      # the dependency AND the value are wired here
orders_pipeline()

load_since(read_watermark()) does two things at once: it makes load_since depend on read_watermark, and it passes the returned watermark in. The value still travels through XCom under the hood, so the same "small values only" rule applies.

Mixing styles is fine — a TaskFlow @task and a PythonOperator in the same DAG can be chained with >>.

Sensors: waiting for something#

A sensor is a task that succeeds only once a condition is met — a file appears, a partition lands, a time passes. Until then it keeps checking.

from airflow.sensors.filesystem import FileSensor
wait_for_export = FileSensor(    task_id="wait_for_export",    filepath="/data/exports/orders_{{ ds }}.csv",    poke_interval=60,        # check every 60s    timeout=60 * 60,         # give up after an hour -> task fails    mode="reschedule",       # free the worker slot between checks)
wait_for_export >> load_export

• mode="reschedule" releases the worker between checks instead of holding a slot for the whole wait — important when several sensors run at once.
• timeout is essential: a sensor with no timeout waits forever and blocks the run.

Common sensors: FileSensor, DateTimeSensor, ExternalTaskSensor (wait for a task in another DAG), and provider sensors like S3KeySensor.

Other operators worth knowing#

For a pipeline that is mostly pandas and SQL in Python, PythonOperator (or @task) covers the large majority of the work.

Writing a retry-safe task#

Every task can run more than once — a retry, a manual re-run, a backfill. A task that is not safe to repeat corrupts data on the second run. Make each task idempotent:

• Replace, do not append. Instead of INSERT into a target, delete the affected partition or key range first, then insert. Re-running rewrites the same rows.
• Use the data interval as the boundary. Process exactly the window the run owns, so two runs never overlap.
• Make external effects safe to repeat. Writing a file: write to a temp name and rename. Calling an API: use an idempotency key if the API supports one.

Idempotent tasks and safe backfills are covered further in Scheduling and Backfill and, on the SQL side, Building Staging and Data Marts.

Keep logic out of the operator#

The python_callable should be a plain function in its own module:

# transform/raw_to_staging.py  — no Airflow importsdef load_raw_orders_to_staging(**context):    ...

# dags/orders_pipeline.pyfrom transform.raw_to_staging import load_raw_orders_to_staging
load_staging = PythonOperator(    task_id="load_raw_orders_to_staging",    python_callable=load_raw_orders_to_staging,)

That function can be imported and run in a test with a fake context, without Airflow running at all. Logic written directly inside the DAG file cannot.

Common mistakes#

datetime.now() inside a task#

Makes the result depend on when the task happens to run. Use data_interval_start / data_interval_end from the context.

Large values in op_kwargs or XCom#

These go through the metadata database. Pass configuration and pointers, not datasets.

A task that appends on every run#

A retry or backfill then double-loads. Delete-then-insert the affected range so re-runs are safe.

Airflow imports in the callable module#

The function should be plain Python so it can be tested standalone. Keep the Airflow wiring in the DAG file.

One giant task that does the whole pipeline#

If load, transform, and publish are one task, a failure in publish re-runs the load. Split work at the points where you would want to retry independently.

Quick reference#

See also#