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.

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.

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.

Troubleshooting matrix #

Related #

• Scaling Celery Pipelines for Bulk Ingestion — the queue-topology rationale this procedure implements.
• Aggregating Celery Logs with Loki — tracing a scaled worker fleet’s logs as one searchable stream.
• Implementing OAI-PMH Resumption Tokens — paging the upstream harvest that feeds these worker queues.

Up one level: Scaling Celery Pipelines for Bulk Ingestion.