What makes a task idempotent?

Running it twice produces the same result as running it once. In practice that means deleting the target partition before writing it, or using an upsert keyed on something stable. Airflow retries tasks and people rerun backfills, so a non-idempotent task will eventually double-count.

When do I need a pool?

When a resource is shared and finite — database connections, an API rate limit, a licensed tool. A pool caps how many tasks can use it concurrently across every DAG, which a per-DAG concurrency setting cannot do.

Why is my scheduler slow?

Usually top-level code in DAG files. The scheduler re-parses every file on an interval, so a database query, an API call, or a heavy import at module level runs constantly. Check the dag_processing.last_duration metric per file.

Should I use SLAs or a timeout?

Both, for different things. execution_timeout kills a task that runs too long. An SLA does not kill anything; it records a miss and sends a notification, which is right when late is a problem but killing the task would be worse.

Airflow in Production: Idempotency, Pools, and SLAs

A DAG that works on your machine and a DAG that runs unattended every night are different artefacts. These are the properties that separate them.

Idempotency

The most important property, and the easiest to get wrong:

    @task
    def load_bad(day: str) -> None:
        hook = PostgresHook(postgres_conn_id="warehouse")
        hook.run("INSERT INTO daily_revenue (day, revenue) VALUES (%s, %s)",
                 parameters=(day, 12840.55))

Run it, let it be retried once, and check:

 day        | revenue
------------+----------
 2026-02-14 | 12840.55
 2026-02-14 | 12840.55

Duplicated. Airflow retries by default, and a backfill reruns deliberately — so this table is wrong the first time anything goes slightly wrong.

Two ways to fix it. Upsert on a key:

    @task
    def load_upsert(day: str) -> None:
        hook = PostgresHook(postgres_conn_id="warehouse")
        hook.run("""
            INSERT INTO daily_revenue (day, revenue) VALUES (%s, %s)
            ON CONFLICT (day) DO UPDATE SET revenue = EXCLUDED.revenue,
                                            updated_at = NOW()
        """, parameters=(day, 12840.55))

Or delete the partition first, in one transaction:

    @task
    def load_replace(day: str) -> None:
        hook = PostgresHook(postgres_conn_id="warehouse")
        hook.run([
            "DELETE FROM daily_revenue WHERE day = %(day)s",
            "INSERT INTO daily_revenue (day, revenue) SELECT %(day)s, SUM(amount) "
            "FROM orders WHERE order_date = %(day)s",
        ], parameters={"day": day}, autocommit=False)

 day        | revenue
------------+----------
 2026-02-14 | 12840.55

Once, however many times you run it. autocommit=False makes both statements one transaction, so a crash between them cannot leave the partition empty.

For files, write to a temporary path and move atomically:

    @task
    def write_parquet(day: str) -> None:
        tmp = f"s3://lake/_tmp/{day}/"
        final = f"s3://lake/revenue/day={day}/"
        df.write.mode("overwrite").parquet(tmp)
        S3Hook("aws_default").copy_object(tmp, final)   # then delete tmp

Writing directly to final means a failure halfway leaves a partial partition that downstream readers will happily consume.

Pools

A backfill of 90 days will happily open 90 database connections at once:

airflow pools set warehouse 5 "Warehouse connection slots"
airflow pools list

Pool        | Slots | Description
============+=======+=============================
default_pool| 128   | Default pool
warehouse   | 5     | Warehouse connection slots

    load = PostgresOperator(
        task_id="load",
        postgres_conn_id="warehouse",
        sql="...",
        pool="warehouse",
        pool_slots=1,
    )

airflow dags backfill revenue --start-date 2025-11-16 --end-date 2026-02-14

[backfill progress] finished run 12 of 90 | tasks waiting: 78 | running: 5 | succeeded: 12

Five at a time regardless of how many runs are queued. Pools are global — every DAG using pool="warehouse" shares the same five slots, which is exactly what a shared database needs and what per-DAG max_active_tasks cannot give you.

A heavy task can claim several slots:

    heavy = SparkSubmitOperator(task_id="heavy", pool="warehouse", pool_slots=3)

Concurrency, in order of scope

with DAG(
    dag_id="controlled",
    max_active_runs=1,        # one run of THIS dag at a time
    max_active_tasks=8,       # concurrent tasks within one run
    default_args={
        "pool": "warehouse",  # global cap across all dags
        "retries": 3,
        "retry_delay": timedelta(minutes=5),
        "retry_exponential_backoff": True,
        "max_retry_delay": timedelta(minutes=30),
        "execution_timeout": timedelta(hours=2),
    },
) as dag:
    ...

max_active_runs=1 matters more than it looks. Without it, a backfill runs many intervals concurrently, and any task that appends to a shared table will interleave in unpredictable order.

retry_exponential_backoff is worth setting whenever the failure might be a rate limit — retrying every 5 minutes against a throttled API just extends the throttling.

Timeouts

    slow = BashOperator(
        task_id="slow",
        bash_command="sleep 600",
        execution_timeout=timedelta(seconds=30),
    )

[2026-02-15 15:04:31] {taskinstance.py:1938} ERROR - Task failed with exception
airflow.exceptions.AirflowTaskTimeout: Timeout, PID: 48211
[2026-02-15 15:04:31] {taskinstance.py:1206} INFO - Marking task as UP_FOR_RETRY

Without execution_timeout, a task that hangs on a network call holds its slot indefinitely. Set it on anything touching an external system.

At DAG level:

with DAG(dag_id="bounded", dagrun_timeout=timedelta(hours=4), ...) as dag:

SLAs

An SLA does not kill the task — it records that the task finished later than promised:

def sla_missed(dag, task_list, blocking_task_list, slas, blocking_tis):
    print(f"SLA MISS on {dag.dag_id}: {[s.task_id for s in slas]}")
    print(f"blocked by: {[t.task_id for t in blocking_tis]}")


with DAG(
    dag_id="with_sla",
    sla_miss_callback=sla_missed,
    default_args={"sla": timedelta(minutes=30)},
    ...
) as dag:
    ...

[2026-02-15 15:32:02] {dag.py:1204} INFO - Calling SLA miss callback
SLA MISS on with_sla: ['transform']
blocked by: ['extract']

The callback names both the task that missed and what was blocking it — usually the more useful of the two. Missed SLAs are also visible under Browse → SLA Misses.

Use execution_timeout when running long is itself a problem, and an SLA when running late is a problem but killing the work would be worse.

Alerting

def on_failure(context):
    ti = context["task_instance"]
    print(f"FAILED {ti.dag_id}.{ti.task_id} on {context['ds']}")
    print(f"try {ti.try_number} of {ti.max_tries + 1}")
    print(f"log: {ti.log_url}")
    # send to Slack, PagerDuty, etc.


with DAG(
    dag_id="alerting",
    default_args={
        "on_failure_callback": on_failure,
        "email_on_failure": False,          # callbacks beat email
        "email_on_retry": False,
    },
    ...
) as dag:
    ...

FAILED alerting.transform on 2026-02-14
try 3 of 3
log: http://airflow.internal/log?dag_id=alerting&task_id=transform&execution_date=...

Note email_on_retry=False. A task with three retries emails three times for one incident, and teams learn to ignore the alerts — which defeats the point.

For a single alert per run rather than per task:

with DAG(dag_id="alerting", on_failure_callback=dag_failed, ...) as dag:

Keeping the scheduler fast

The scheduler re-parses every DAG file on min_file_process_interval (30s default). Anything at module level runs on that cadence:

# BAD — runs every 30 seconds, forever, per file
import pandas as pd
CONFIG = requests.get("https://config.internal/pipeline").json()
TABLES = PostgresHook("warehouse").get_records("SELECT name FROM tables")

with DAG(...) as dag:
    for t in TABLES:
        ...

airflow dags report

file                        | duration | dag_num | task_num
============================+==========+=========+==========
/dags/bad_toplevel.py       | 4.82s    | 1       | 12
/dags/daily_summary.py      | 0.04s    | 1       | 3
/dags/revenue.py            | 0.03s    | 1       | 5

4.82 seconds against 0.04. With 50 such files the scheduler spends all its time parsing and tasks queue for minutes before starting.

Move the work into tasks:

# GOOD — parse is instant; the work happens when the task runs
with DAG(...) as dag:

    @task
    def get_tables() -> list[str]:
        return [r[0] for r in PostgresHook("warehouse").get_records("SELECT name FROM tables")]

    process.expand(table=get_tables())

/dags/good_toplevel.py      | 0.05s    | 1       | 3

Also worth setting:

[scheduler]
min_file_process_interval = 60
dag_dir_list_interval = 120
parsing_processes = 4

Log retention

Task logs and metadata rows accumulate forever by default:

airflow db clean --clean-before-timestamp '2025-11-01' --yes

Performing db clean before 2025-11-01T00:00:00+00:00
  task_instance: 1,284,021 rows to delete
  dag_run: 48,201 rows to delete
  log: 3,910,447 rows to delete
  xcom: 291,004 rows to delete
Deleted 5,533,673 rows

An Airflow database nobody cleans becomes the reason the UI is slow. Schedule this monthly.

A production checklist

Practice

1. Write a plain INSERT task, let it retry, and count the rows.

 2026-02-14 | 12840.55
 2026-02-14 | 12840.55

Two rows from one logical run. Switch to ON CONFLICT DO UPDATE and re-run as many times as you like — still one row. This is the single most valuable habit in orchestration.

2. Create a pool of 2 and backfill 30 days.

[backfill progress] finished run 4 of 30 | tasks waiting: 26 | running: 2 | succeeded: 4

Two at a time. Without the pool all 30 start at once and the database refuses connections — which then fails tasks for a reason that looks nothing like “too much concurrency”.

3. Put an HTTP call at module level and check airflow dags report.

/dags/with_http.py    | 4.82s | 1 | 12

Nearly five seconds per parse, every 30 seconds. Move it into a task and the file parses in milliseconds. airflow dags report is the fastest way to find the file slowing your scheduler.

4. Set an SLA of 1 minute on a task that takes 2 minutes.

SLA MISS on with_sla: ['slow_task']

The task still succeeds — an SLA only records and notifies. If you need it killed, use execution_timeout; if you need to know it was late, use both.

That completes the Airflow track: from a first DAG through XComs, scheduling, dynamic mapping, sensors, branching, connections, testing, and production hardening.