Scaling Celery Pipelines for Bulk Metadata Ingestion

When a portal harvests tens of thousands of records, regenerates thumbnails, and reindexes in one pass, a single default Celery queue becomes the bottleneck: bulk jobs saturate every worker and interactive tasks — a user’s publish, a single-record edit — wait behind them. GIS administrators see the portal “hang” during a harvest window, and platform engineers get paged for broker backlogs that never drain. This guide sits within the Metadata Catalog Automation & Ingestion Workflows practice and treats the task queue as an engineered system with its own topology, capacity model, and backpressure controls rather than a single pool that absorbs whatever is thrown at it.

The core failure is workload interference. Bulk harvests, thumbnail rasterization, and search reindexing have completely different resource profiles and latency tolerances, yet the naive deployment routes them all to one queue served by one worker pool. The fix is a deliberate queue topology, per-workload concurrency, autoscaling driven by broker depth, and rate limits that protect PostGIS and the search index downstream.

Architectural Placement: The Queue as a Capacity Boundary #

The task queue sits between the ingestion triggers — a scheduled harvest, an operator-initiated reindex, a user publishing a layer — and the shared stores that every one of those actions eventually writes to. Its job is not merely to defer work but to isolate workloads from each other so that a 50,000-record harvest cannot evict a user’s single publish from the runnable set. That isolation is impossible with one queue and one pool; it requires a topology where each workload has a named queue, a dedicated worker pool, and a capacity budget it cannot exceed.

Bulk ingestion in a geospatial portal is not one workload but several with sharply different profiles. Harvesting is I/O-bound — mostly waiting on remote HTTP endpoints and the PostGIS write path — so a single process can multiplex hundreds of concurrent tasks cheaply with a gevent or eventlet pool. Thumbnail generation is CPU-bound rasterization that must run under the prefork pool with concurrency near the core count, or it will thrash. Reindexing is bursty and best done in batches that hit the search index with bulk requests rather than one document per task. Routing all three to the same queue guarantees head-of-line blocking; separating them lets each scale on its own signal.

The records this pipeline processes arrive from the harvest layer. Where those records are pulled incrementally rather than in full sweeps, incremental metadata harvesting and change detection governs which records enter the queue in the first place, and automated metadata ingestion via OAI-PMH supplies the standardized pull mechanism upstream of every task described here. This guide assumes the records are already selected and focuses entirely on draining them through workers without collapsing the portal.

Queue Topology and Task Routing #

Give every workload a dedicated queue and route tasks to it declaratively with task_routes. Routing by task name — not by ad-hoc apply_async(queue=...) calls scattered through the codebase — keeps the topology in one auditable place and lets you re-point a workload without touching call sites.

# celeryconfig.py — queue topology and per-workload routing.
# Every task is pinned to a queue by name; interactive work never shares a
# queue with a bulk harvest, so a large sweep cannot starve a user's publish.
from kombu import Queue

task_queues = (
    Queue("interactive", routing_key="interactive"),   # user-facing, keep drained
    Queue("harvest",     routing_key="harvest"),        # bulk, I/O-bound
    Queue("thumbnails",  routing_key="thumbnails"),     # CPU-bound rasterization
    Queue("reindex",     routing_key="reindex"),        # bursty, batched writes
    Queue("dead_letter", routing_key="dead_letter"),    # exhausted retries land here
)

task_default_queue = "interactive"
task_routes = {
    "catalog.tasks.publish_layer":     {"queue": "interactive"},
    "catalog.tasks.harvest_record":    {"queue": "harvest"},
    "catalog.tasks.render_thumbnail":  {"queue": "thumbnails"},
    "catalog.tasks.reindex_batch":     {"queue": "reindex"},
}

# Fair dispatch: don't let one worker hoard a backlog it cannot start soon.
worker_prefetch_multiplier = 1
task_acks_late = True                 # ack only after success; redelivery on crash
task_reject_on_worker_lost = True     # requeue if the worker dies mid-task

Priority separation is structural here, not a numeric priority field. Because the interactive queue is served by its own always-on pool, a flood on harvest or reindex physically cannot occupy the workers that drain user-facing work. If your broker is RabbitMQ you can additionally enable per-message priorities within a queue, but the primary isolation is the dedicated-queue boundary, which behaves identically on Redis and RabbitMQ.

Worker Concurrency, Prefetch, and Long-Task Semantics #

Concurrency and prefetch are the two levers that decide whether a pool runs efficiently or corrupts its own throughput. Match the pool type to the workload: --pool gevent (or eventlet) with high -c for I/O-bound harvest tasks, --pool prefork with -c near the core count for CPU-bound thumbnails. The prefetch multiplier controls how many messages a worker reserves beyond those it is actively running; the default of 4 is wrong for long tasks because it lets one worker grab a batch it will hold for minutes while other idle workers see an empty queue.

# Harvest pool: I/O-bound, high concurrency, minimal prefetch so long tasks
# stay evenly distributed instead of piling onto one greedy worker.
celery -A catalog worker \
  --queues harvest \
  --pool gevent --concurrency 100 \
  --prefetch-multiplier 1 \
  --max-tasks-per-child 500 \
  --hostname harvest@%h

# Thumbnail pool: CPU-bound, concurrency near core count, prefork isolation.
celery -A catalog worker \
  --queues thumbnails \
  --pool prefork --concurrency 4 \
  --prefetch-multiplier 1 \
  --max-memory-per-child 512000 \
  --hostname thumb@%h

Long tasks demand acks_late = True so a message is acknowledged only after the task succeeds; if the worker is killed mid-harvest — an OOM, a node drain, a rescheduled pod — the broker redelivers the message rather than silently losing the record. Pair it with a visibility_timeout on Redis that comfortably exceeds your longest task, or the broker will redeliver a still-running task to a second worker and you will process it twice. On RabbitMQ the equivalent guard is a consumer acknowledgement timeout set above the task’s worst-case duration. Always set explicit task_time_limit and task_soft_time_limit so a hung remote endpoint cannot pin a worker slot forever — the soft limit raises a catchable exception for cleanup, the hard limit kills the task.

Autoscaling on Broker Queue Depth #

CPU-based autoscaling is the wrong signal for a queue worker. A harvest worker waiting on a slow remote endpoint sits near-idle on CPU while its queue backs up by thousands of messages; scaling on CPU would never add capacity when it is most needed. Scale on the metric that actually reflects pending work — the broker queue length — using KEDA, which reads Redis list length or RabbitMQ queue depth directly and drives the underlying Deployment’s replica count.

# keda-harvest-scaledobject.yaml — scale harvest workers on Redis queue length.
# Plain YAML with no templating, so it validates and applies as-is.
apiVersion: keda.sh/v1alpha1
kind: ScaledObject
metadata:
  name: harvest-worker
  namespace: geoportal
spec:
  scaleTargetRef:
    name: celery-harvest-worker      # the Deployment running the harvest pool
  minReplicaCount: 1                  # keep one warm to avoid cold-start latency
  maxReplicaCount: 20
  cooldownPeriod: 120                 # wait before scaling back to min
  triggers:
    - type: redis
      metadata:
        address: redis.geoportal.svc.cluster.local:6379
        listName: harvest             # the Celery queue name = Redis list key
        listLength: "50"              # target ~50 pending tasks per replica
        activationListLength: "5"     # scale from zero only past this backlog

KEDA’s advantage over a raw Horizontal Pod Autoscaler is scale-to-and-from-zero and native broker triggers; behind the scenes it still creates an HPA, so the two are not exclusive. Keep minReplicaCount at 1 for interactive and harvest pools to avoid cold-start latency on the first task, but let reindex scale from zero since it runs in bursts. The KEDA project documents the full trigger set at keda.sh, and the Celery routing and worker options above are specified in the Celery documentation.

Idempotency, Retries, and Dead-Letter Handling #

Autoscaling and acks_late together guarantee at-least-once delivery, which means every task will occasionally run twice — on redelivery after a crash, or when a visibility timeout fires. Bulk ingestion is only safe under that guarantee if each task is idempotent: writing the same record twice must converge to the same state, never duplicate it. Key every write on a stable record identifier and upsert rather than insert, so a replayed harvest_record overwrites cleanly instead of creating a shadow duplicate in PostGIS or the index.

# catalog/tasks.py — idempotent harvest with bounded retry and dead-lettering.
from celery import shared_task
from celery.utils.log import get_task_logger

log = get_task_logger(__name__)

@shared_task(
    bind=True,
    acks_late=True,
    max_retries=5,
    autoretry_for=(ConnectionError, TimeoutError),
    retry_backoff=2,          # 2s, 4s, 8s, 16s, 32s exponential backoff
    retry_backoff_max=300,
    retry_jitter=True,        # spread retries so they don't thundering-herd
    time_limit=600,
    soft_time_limit=540,
)
def harvest_record(self, record_id: str, payload: dict) -> None:
    try:
        # Upsert keyed on record_id -> a redelivered task is a no-op overwrite.
        upsert_record(record_id, payload)
    except PermanentValidationError as exc:
        # Do not retry a record that will never pass; send it to the DLQ.
        log.error("dead-lettering %s: %s", record_id, exc)
        dead_letter.apply_async((record_id, payload, str(exc)), queue="dead_letter")
        return

Distinguish transient from permanent failures. Transient errors — a remote timeout, a broker blip — deserve exponential backoff with jitter, bounded by max_retries. Permanent errors — a record that fails validation and always will — must not consume retry budget; route them immediately to the dead_letter queue with the original payload and the failure reason, exactly as the ingestion gate quarantines malformed metadata upstream. A task that has exhausted its retries should also land in the dead-letter queue rather than vanishing, so an operator can inspect, fix the source, and replay in bulk.

Backpressure and Protecting Downstream Stores #

The whole point of scaling workers is throughput, but unbounded throughput is a weapon aimed at PostGIS and the search index. Twenty harvest workers at concurrency 100 can open 2,000 concurrent write attempts; PostgreSQL’s connection ceiling is a few hundred, and blowing past it produces FATAL: too many connections long before the pooler saturates. Backpressure is the discipline that keeps worker fan-out from exceeding what the stores can absorb.

Apply it at two levels. Per-task, Celery’s rate_limit caps the dispatch rate of a task type — rate_limit="200/m" on reindex_batch bounds how fast batches hit the index regardless of how many workers exist. Structurally, front PostGIS with a connection pooler so worker concurrency is decoupled from database connections: a hundred workers share a bounded pool of a few dozen server connections, and the pooler queues the overflow instead of the database rejecting it. Batch writes rather than issuing one statement per record — a reindex_batch task that bulk-indexes 500 documents in one request is dramatically cheaper on the index than 500 single-document tasks.

Chunk large harvests at the producer. Rather than enqueuing 50,000 individual harvest_record tasks in one loop — which floods the broker and can exhaust its memory — split the identifier list into chunks and enqueue them progressively, letting the queue-depth autoscaler pace ingestion against real drain rate. This is where the companion page goes deep: scaling Celery workers for bulk metadata ingestion walks through the concrete worker, KEDA, and chunking configuration end to end with verification steps.

Operational Troubleshooting #

Diagnose ingestion-pipeline failures by the symptom operators actually observe, the queue or metric that reveals the cause, and the control surface that fixes it:

Instrument every pool. Export per-queue depth, task success and failure rate, retry counts, and end-to-end latency so the autoscaler and the on-call engineer see the same picture. Centralizing the worker logs makes a failed harvest traceable across every replica at once — aggregating Celery logs with Loki is the mechanism that turns twenty ephemeral worker pods into one searchable stream keyed by task and record id.

Related #

• Metadata Catalog Automation & Ingestion Workflows — the parent reference this scaling guide fits inside.
• Scaling Celery Workers for Bulk Metadata Ingestion — the hands-on worker, KEDA, and chunking configuration behind this topology.
• Automated Metadata Ingestion via OAI-PMH — the standardized pull mechanism that feeds records into these queues.
• Incremental Metadata Harvesting and Change Detection — how records are selected before they ever enter the pipeline.
• Aggregating Celery Logs with Loki — making a distributed worker fleet’s logs searchable as one stream.

Up one level: Metadata Catalog Automation & Ingestion Workflows.