Skip to Content
AirflowWorker Queues

Worker Queues

Worker queues let one Airflow instance run multiple pools of Celery workers, each with its own machine size, node pool, concurrency, and autoscaling range. DAG authors route tasks to a queue with queue="name" on the operator; everything else keeps running on the default queue.

If you’ve used Astronomer’s worker queues, this is the same model.

When to use a queue

Every Airflow instance starts with a single default queue. Add a named queue when a subset of your tasks needs a different trade-off than the rest:

  • Protect long-running tasks from spot preemption. If your default workers run on a spot node pool, a 30-minute dbt build or dlt extraction can get killed mid-run every time the cloud reclaims an instance — and exhaust its retries without ever failing “for real”. Define a stable queue on an on-demand pool and route only the critical tasks there; the cheap fan-out work stays on spot.
  • Memory-hungry tasks. A highmem queue with 8 GiB workers and low concurrency keeps a handful of heavy tasks from OOMing workers sized for the other 95% of your DAGs.
  • Scale-to-zero for expensive shapes. Queues can set Min Workers to 0 — the KEDA autoscaler only spins workers up while tasks are queued, so a big machine shape costs nothing while idle.

Defining a queue

In the console, open your Airflow application and go to the Worker Queues tab:

  1. Click + Add Queue.
  2. Give it a name — 1–32 lowercase letters, digits, and hyphens (highmem, stable, bulk-sync). The name is also the Celery queue tasks route to.
  3. Pick a node pool. Spot pools automatically get the spot toleration, same as the default queue.
  4. Set CPU / Memory (per worker, request = limit), Concurrency (tasks per worker), and the Min / Max Workers autoscaling range.
  5. Save & Apply.

Each queue becomes its own worker Deployment (airflow-worker-<name>; once any queue exists the default queue’s Deployment is named airflow-worker-default-queue) with its own KEDA autoscaler that watches only that queue’s queued + running tasks.

The default queue is shown as the first row. You can resize and place it like any other queue, but not delete it — tasks that don’t name a queue have to land somewhere. (Environment presets on the Execution tab also set the default queue’s size as part of the overall environment shape.)

Removing a queue removes its Deployment on the next apply. Running tasks get their normal termination grace period to finish; anything still queued on a removed queue will sit until a queue with that name exists again, so re-route the DAGs first.

Routing tasks (CeleryExecutor)

Set queue on any operator:

from airflow.providers.standard.operators.python import PythonOperator
 
heavy = PythonOperator(
    task_id="build_features",
    python_callable=build_features,
    queue="highmem",
)

Or for a whole DAG via default_args:

with DAG(
    dag_id="dbt_nightly",
    default_args={"queue": "stable"},
    ...
):
    ...

Tasks without a queue run on the default queue.

Concrete example — Cosmos dbt on spot: a Cosmos DbtDag watcher task lives for the whole dbt build. On a spot pool it can be preempted 2–3 times in a row and burn all its retries. Route it to an on-demand queue:

DbtDag(
    dag_id="dbt_prod",
    operator_args={"queue": "stable"},  # on-demand pool — preemption-proof
    ...
)

Routing tasks (KubernetesExecutor)

KubernetesExecutor tasks don’t use Celery queues — each task already gets its own pod (see Executors for when to use which). To place a KE task (including tasks opted into KE from hybrid mode) on a specific node pool, override the pod spec with executor_config:

from kubernetes.client import models as k8s
 
place_on_highmem = {
    "pod_override": k8s.V1Pod(
        spec=k8s.V1PodSpec(
            containers=[
                k8s.V1Container(
                    name="base",
                    resources=k8s.V1ResourceRequirements(
                        requests={"cpu": "2", "memory": "8Gi"},
                        limits={"cpu": "2", "memory": "8Gi"},
                    ),
                )
            ],
            node_selector={
                # GKE: cloud.google.com/gke-nodepool
                # EKS / AKS: node-pool
                "cloud.google.com/gke-nodepool": "highmem-pool"
            },
        )
    )
}
 
heavy = PythonOperator(
    task_id="build_features",
    python_callable=build_features,
    executor="KubernetesExecutor",  # hybrid mode only; omit on KE-only
    executor_config=place_on_highmem,
)

If the target pool is a spot pool, also add the spot toleration to the pod override (key: spot, value: "true", effect NoSchedule).

How autoscaling works

Each queue scales independently: the autoscaler polls the Airflow metadata database and targets

workers = ceil((queued + running tasks on this queue) / concurrency)

clamped to the queue’s Min/Max range. A queue’s concurrency is enforced on its workers directly, so a highmem queue with concurrency 4 really runs at most 4 tasks per worker even if the instance-wide default is 16.

Limits

  • Queues are per-application; two Airflow instances don’t share queues.
  • Queue names must be unique within an instance, and default is reserved.
  • If a queue references a node pool that has since been deleted, the config still applies (with a warning in the logs) but that queue’s workers stay Pending until the pool exists again or you repoint the queue.
Last updated on
ExecutorsOverview