Geospatial Portal

Scaling Celery Workers for Bulk Metadata Ingestion

A concrete procedure for configuring Celery workers, KEDA queue-depth autoscaling, and chunked enqueueing so a bulk metadata harvest drains at high throughput without starving interactive tasks or overwhelming PostGIS.

This guide is a hands-on companion to Scaling Celery Pipelines for Bulk Ingestion and sits within the wider Metadata Catalog Automation & Ingestion Workflows practice; read the parent guide first for the queue-topology rationale. Here the focus is narrow and operational: the exact worker flags, the KEDA ScaledObject, the acks and time limits, the chunked batch writes, and the rate limits that keep a large harvest safe — plus how to verify each one took effect.

Prerequisites

Confirm the following before applying any configuration. Each is a common cause of a harvest that either crawls or takes the portal down with it.

  • Celery 5.3+ with a broker that exposes queue depth: Redis 7+ (list length) or RabbitMQ 3.12+ (queue messages).
  • A running KEDA 2.13+ in the Kubernetes cluster (keda-operator and keda-operator-metrics-apiserver healthy in the keda namespace).
  • The worker Deployments already exist — one per pool (celery-harvest-worker, etc.) — so KEDA has a scaleTargetRef to drive.
  • kubectl access to the geoportal namespace and permission to create ScaledObject custom resources.
  • PgBouncer (or an equivalent pooler) fronting PostGIS, so worker concurrency is decoupled from the database connection ceiling.
  • Harvest tasks are idempotent (upsert on a stable record id) — mandatory before autoscaling introduces at-least-once redelivery.

The control loop below is what you are building: pending work in the broker drives the autoscaler, which adds workers, which drain the queue, which lowers depth — a closed loop that paces ingestion against real capacity.

Queue-depth autoscaling control loop for a bulk harvest A chunked producer enqueues harvest tasks in batches into a broker queue, shown as a cylinder with a live depth reading. KEDA polls the queue depth and compares it to the target of about fifty pending tasks per replica. When depth exceeds the target it increases the worker Deployment replica count up to the maximum; the added workers consume tasks and drain the queue, which lowers depth back toward the target. After a cooldown with depth below target, KEDA scales the Deployment back down toward its minimum. Arrows form a closed loop from queue depth to scaler to workers and back to queue depth. Chunked producer enqueue in batches of 500 ids Broker queue depth = N pending KEDA scaler compare depth to target 50 / replica Worker Deployment gevent -c 100 replicas 1..20 acks_late · prefetch 1 poll depth set replicas drain tasks

Step-by-step implementation

1. Define dedicated queues and route bulk tasks

Pin bulk tasks to their own queue so they never compete with interactive work for a worker slot. Declare the routing once, in config, keyed on task name.

# celeryconfig.py — route bulk harvest work onto its own queue.
from kombu import Queue

task_queues = (
    Queue("interactive", routing_key="interactive"),
    Queue("harvest",     routing_key="harvest"),
    Queue("dead_letter", routing_key="dead_letter"),
)
task_default_queue = "interactive"
task_routes = {
    "catalog.tasks.harvest_record": {"queue": "harvest"},
}

2. Tune worker concurrency and prefetch for I/O-bound harvests

Harvest tasks spend most of their time waiting on remote endpoints and the write path, so run them under a gevent pool with high concurrency and a prefetch multiplier of 1 — the low prefetch keeps long tasks spread evenly instead of piling onto one worker while others idle.

# Harvest worker: I/O-bound, high concurrency, fair dispatch.
celery -A catalog worker \
  --queues harvest \
  --pool gevent --concurrency 100 \
  --prefetch-multiplier 1 \
  --max-tasks-per-child 500 \
  --hostname harvest@%h \
  --loglevel INFO

--max-tasks-per-child 500 recycles each worker process after 500 tasks, which caps the slow memory growth that long harvest runs otherwise accumulate.

3. Deploy a KEDA ScaledObject that scales on queue length

KEDA reads broker depth directly and drives the worker Deployment’s replicas. Target roughly 50 pending tasks per replica so throughput rises with backlog and settles when the queue drains.

# harvest-scaledobject.yaml — scale harvest workers on Redis queue length.
apiVersion: keda.sh/v1alpha1
kind: ScaledObject
metadata:
  name: harvest-worker
  namespace: geoportal
spec:
  scaleTargetRef:
    name: celery-harvest-worker
  minReplicaCount: 1
  maxReplicaCount: 20
  pollingInterval: 15
  cooldownPeriod: 120
  triggers:
    - type: redis
      metadata:
        address: redis.geoportal.svc.cluster.local:6379
        listName: harvest
        listLength: "50"
        activationListLength: "5"

For a RabbitMQ broker, swap the trigger to type: rabbitmq with queueName: harvest, mode: QueueLength, and value: "50". The full scaler reference is at keda.sh.

4. Set acks_late and task time limits

With workers scaling in and out, a worker will be killed mid-task on scale-down or a node drain. acks_late guarantees the message is redelivered instead of lost; explicit time limits stop a hung remote endpoint from pinning a worker slot forever.

# catalog/tasks.py — durable, bounded harvest task.
from celery import shared_task

@shared_task(
    bind=True,
    acks_late=True,               # ack only after success -> redeliver on crash
    reject_on_worker_lost=True,   # requeue if the worker dies mid-task
    max_retries=5,
    retry_backoff=2,              # 2,4,8,16,32s exponential backoff
    retry_jitter=True,
    time_limit=600,               # hard kill at 10 min
    soft_time_limit=540,          # catchable exception at 9 min for cleanup
)
def harvest_record(self, record_id: str, payload: dict) -> None:
    upsert_record(record_id, payload)   # idempotent: safe to redeliver

On Redis, set the broker visibility_timeout above time_limit (e.g. 900s) so the broker does not redeliver a still-running task to a second worker.

5. Batch DB writes in chunks

Do not enqueue 50,000 individual tasks in one loop — that floods the broker. Split the id list into chunks and enqueue progressively, letting the queue-depth autoscaler pace ingestion against real drain rate.

# catalog/producer.py — chunked enqueue so the broker is never flooded.
from itertools import islice
from catalog.tasks import harvest_record

def chunked(iterable, size):
    it = iter(iterable)
    while batch := list(islice(it, size)):
        yield batch

def enqueue_harvest(record_ids, payloads, chunk_size=500):
    total = 0
    for batch in chunked(record_ids, chunk_size):
        for rid in batch:
            harvest_record.apply_async((rid, payloads[rid]), queue="harvest")
        total += len(batch)
    return total   # log the count; queue depth now drives autoscaling

Where the write itself can be batched — a bulk index request or a COPY-style multi-row insert — collapse a chunk into a single downstream call rather than one statement per record.

6. Cap throughput with rate_limit to protect downstreams

Unbounded worker fan-out can open more concurrent writes than PostGIS or the index can absorb. A per-task rate_limit bounds dispatch regardless of replica count, giving the pooler and the index headroom.

# Bound how fast harvest tasks dispatch per worker, independent of scale.
@shared_task(rate_limit="200/m", acks_late=True)
def harvest_record(self, record_id: str, payload: dict) -> None:
    upsert_record(record_id, payload)

Combined with a bounded PgBouncer pool, this keeps a 20-replica fleet from ever presenting more connections than PostGIS is configured to accept.

Concurrency, prefetch, and why a fast worker can look idle

Two settings determine how work actually reaches a worker, and they interact in a way that produces the most confusing symptom in a scaled pipeline: a queue with thousands of messages, workers reporting low CPU, and throughput far below what the fleet should manage. The cause is almost always prefetch — each worker has already reserved a batch of messages, so the queue looks busy while individual workers sit blocked on one slow task holding a reservation that other, idle workers cannot take.

Prefetch decides whether idle workers can help Two scenarios: high prefetch with reserved messages stranded behind a slow task, and prefetch of one with work distributed as capacity frees. High prefetch — messages reserved in advance worker A 1 slow task + 15 reserved worker B idle — nothing left to take queue depth stays high, CPU stays low, and adding workers changes nothing Prefetch of one — taken as capacity frees worker A 1 slow task, nothing held worker B working through the rest a slow task delays only itself; the fleet stays busy and scaling has an effect

The rule of thumb is to set prefetch by task-duration variance rather than by throughput ambition. Uniform sub-second tasks benefit from batching, because the round trip to the broker dominates. Ingestion tasks whose duration ranges from milliseconds to minutes — which describes almost every metadata pipeline, because record sizes and upstream response times vary enormously — should take one at a time and let the broker do the distribution.

Verification

Confirm each layer took effect before trusting a production harvest.

# 1. Inspect live queue depth on Redis (the list backing the harvest queue)
kubectl exec -n geoportal redis-0 -- redis-cli LLEN harvest
#   (integer) 4820        # pending harvest tasks right now

# 2. Confirm the ScaledObject is active and reporting the trigger
kubectl get scaledobject harvest-worker -n geoportal
#   NAME             SCALETARGETKIND   MIN   MAX   READY   ACTIVE
#   harvest-worker   Deployment        1     20    True    True

# 3. Watch replicas scale up as depth exceeds the per-replica target
kubectl get hpa -n geoportal -w
#   keda-hpa-harvest-worker   ...   TARGETS   50/50   MINPODS 1   REPLICAS 1->12

# 4. Confirm workers registered and are draining their queue
celery -A catalog inspect active_queues | grep harvest
celery -A catalog inspect stats | grep -E 'total|pool'

# 5. Confirm throughput: queue depth should fall over successive polls
watch -n 5 'kubectl exec -n geoportal redis-0 -- redis-cli LLEN harvest'

A healthy run shows LLEN harvest climbing when the producer enqueues, REPLICAS rising toward the max while depth exceeds target, then depth falling and replicas scaling back after the cooldown once the backlog clears.

Scaling down without losing work

Scale-up is forgiving; scale-down is where a bulk run loses records. A worker removed mid-task takes its in-flight work with it, and whether that work is recovered depends entirely on when the message was acknowledged. Acknowledging on receipt makes the queue look tidy and loses the task; acknowledging on completion means a killed worker’s message returns to the queue and is retried elsewhere, which is what you want — provided the task is idempotent, because the message may be redelivered after the work was in fact partly done.

Acknowledgement timing decides what a scale-down costs Two rows comparing acknowledge-on-receipt and acknowledge-on-completion when a worker is terminated mid-task. Acknowledge on receipt the message is gone before the work starts worker terminated → task lost, silently the run finishes with fewer records and no error Acknowledge on completion the message stays reserved until the task ends requires an idempotent task worker terminated → message redelivered elsewhere partial work may repeat, which the upsert absorbs

Pair late acknowledgement with a termination grace period longer than the p99 task duration, so an orderly scale-down usually finishes its work rather than relying on redelivery. Redelivery is the safety net, not the plan — and a fleet that depends on it during every scale-down will eventually meet the task that is not as idempotent as everyone believed.

Troubleshooting matrix

Symptom Likely cause Fix
Queue depth climbs, replicas stay at min ScaledObject not ACTIVE, or wrong listName Check kubectl describe scaledobject; match listName to the Celery queue name exactly
Replicas scale but throughput flat Prefetch too high, one worker hoarding the backlog Set --prefetch-multiplier 1; confirm gevent pool for I/O-bound tasks
Tasks run twice, duplicate records Redelivery under acks_late without idempotency, or low visibility timeout Upsert on a stable id; raise Redis visibility_timeout above time_limit
FATAL: too many connections on PostGIS Worker fan-out exceeds DB ceiling Front PostGIS with PgBouncer; apply rate_limit; bound the pool size
Broker memory spikes at harvest kickoff 50k tasks enqueued at once Enqueue in chunks; let queue-depth autoscaling pace ingestion
Replicas never scale back to min Cooldown too long, or residual depth above activationListLength Lower cooldownPeriod; confirm the queue actually drains to zero
Workers killed mid-task, records lost acks_late unset, so messages ack before completion Set acks_late=True and reject_on_worker_lost=True; verify redelivery

Up one level: Scaling Celery Pipelines for Bulk Ingestion.