Batch Queue Orchestration for Map Rendering at Scale

Rendering a single map sheet is a solved problem; rendering fifty thousand of them overnight, on a fixed pool of workers, without exhausting memory or silently dropping sheets, is an orchestration problem. Once a cartographic pipeline crosses from “export one map” to “produce a national atlas on a schedule,” the bottleneck stops being the renderer and becomes the queue: how work fans out to workers, how failures are retried without corrupting output, how memory pressure is bounded, and how you know — mid-run — how many sheets are done. This guide builds a Celery-based orchestration layer that treats each sheet as an idempotent, retryable unit of work and fans thousands of them out safely. It goes deeper than the basic group/chord example on the Print-Ready Export and Batch Generation Workflows overview, focusing on the production concerns that only surface at scale.

Fan-Out / Fan-In Queue Architecture

Prerequisites and Environment Configuration

Batch orchestration introduces a distributed-systems failure surface on top of the rendering stack. Pin the queue layer explicitly — Celery’s serialization defaults, prefetch behaviour, and result-backend semantics have all shifted between minor versions.

• Python 3.10+ with venv or conda isolation
• celery>=5.3 — required for the stable chord semantics, worker_max_tasks_per_child, and typed autoretry_for behaviour used below
• A broker: redis>=7.0 (with redis-py>=5.0) or RabbitMQ 3.12+ via amqp. Redis is simpler to operate; RabbitMQ gives you real dead-letter exchanges and per-message TTL
• flower>=2.0 — live worker, queue-depth, and task-state monitoring
• psutil>=5.9 — measuring per-render resident memory so concurrency can be derived, not guessed
• The full rendering stack from your export pipeline (matplotlib/Agg, geopandas>=0.14, rasterio>=1.3) — orchestration wraps rendering, it does not replace it
• CRS requirement: every sheet in the manifest must already carry an explicit projected CRS (e.g. EPSG:32617). Resolve projections once, upstream, when the manifest is built — reprojecting inside a task multiplies pyproj initialisation cost across every worker and every retry
• Shared object store: an S3-compatible bucket or a shared POSIX mount reachable from every worker. Workers must never write to node-local disk, or the chord callback cannot assemble the atlas

The render task itself should already be hardened for headless execution and correct output resolution — the DPI and Resolution Management guide covers the backend configuration each worker inherits, and High-Resolution Vector Export covers the vector paths for sheets that must scale without a pixel grid.

Conceptual Foundation

Four ideas separate a batch pipeline that survives production from one that quietly loses sheets.

1. Fan-out / fan-in is a group wrapped by a chord. A group is a parallel set of independent tasks — the fan-out. A chord is a group plus a callback that fires exactly once, after every task in the group has succeeded — the fan-in. The callback is the natural home for cross-sheet work: validating that all sheets rendered, stitching them into a multi-page atlas, or writing a manifest of outputs. The mental model is a barrier: N renders proceed independently, then synchronise at a single gate.

2. Memory, not CPU, bounds render concurrency. A worker process’s job is to hold a full-resolution raster canvas plus intermediate compositing buffers in RAM. At 300 DPI a single letter-size sheet can hold 1–4 GB resident during compositing. This is the opposite of a CPU-bound workload: the limiting resource is the resident set size (RSS) summed across concurrent renders, not core count. Setting --concurrency to the CPU count — the reflex from web-request workloads — oversubscribes memory and invites the kernel OOM-killer. The correct cap is usable_RAM / peak_render_RSS, and it is often lower than the core count.

3. At-least-once delivery demands idempotency. With acks_late enabled (which you want, so a crashed worker’s task is redelivered rather than lost), Celery guarantees at-least-once execution. A task can therefore run twice: once on a worker that crashed after rendering but before acking, and again on redelivery. If the task appends a file or increments a counter, duplicate execution corrupts the batch. The defence is an idempotency key: derive a deterministic output path from a content hash of the rendering inputs, so a re-run overwrites the identical object and the batch converges regardless of how many times any sheet ran.

4. Dead-letter routing makes failure inspectable. Some sheets will fail permanently — a corrupt source geometry, a missing tile, a style that references a deleted layer. Retrying these forever wastes workers; dropping them silently loses data. Dead-letter routing sends a task that has exhausted its retries to a dedicated queue, preserving the sheet spec and the traceback for inspection and later replay, so a batch of 5,000 that has 3 bad sheets still delivers 4,997 and tells you precisely which three to fix.

Step-by-Step Implementation

Step 1: Define an Idempotent Render Task with a Content-Hash Key

The output path is a pure function of the inputs. Hash the parameters that determine the pixels — bbox, layer set, style version, DPI — and write to that key. Retried or duplicated executions land on the same object.

import hashlib
import json

def render_output_key(sheet: dict) -> str:
    """
    Deterministic object key derived from everything that affects the pixels.
    Two invocations with identical inputs yield the identical key, so an
    at-least-once retry overwrites rather than duplicates.
    """
    fingerprint = {
        "bbox": [round(c, 6) for c in sheet["bbox"]],  # (minx, miny, maxx, maxy)
        "crs": sheet["crs"],
        "layers": sorted(sheet["layers"]),
        "style_version": sheet["style_version"],
        "dpi": sheet["dpi"],
    }
    blob = json.dumps(fingerprint, sort_keys=True, separators=(",", ":"))
    digest = hashlib.sha256(blob.encode("utf-8")).hexdigest()[:16]
    return f"sheets/{sheet['atlas_id']}/{sheet['sheet_id']}-{digest}.tiff"

Rounding the bbox to a fixed precision keeps floating-point noise from producing a different hash for what is cartographically the same extent. Sorting layers makes the key insensitive to manifest ordering.

Step 2: Set Memory-Aware Worker Concurrency

Do not guess concurrency — measure one render’s peak RSS with psutil, then derive the cap. Reserve headroom for the OS and the broker client.

import psutil

def safe_concurrency(peak_render_mib: float, reserve_mib: float = 2048.0) -> int:
    """
    Cap concurrency at (usable RAM) / (peak RSS per render).
    peak_render_mib is measured empirically from one render, not assumed.
    """
    usable_mib = (psutil.virtual_memory().total / 1024**2) - reserve_mib
    workers = int(usable_mib // peak_render_mib)
    return max(1, workers)

# Example: 16 GB box, ~3.2 GB peak per 300 DPI sheet
# → (16384 - 2048) // 3200 = 4 concurrent renders, NOT 8 (the core count)

Launch the worker with the derived value: celery -A tasks worker --concurrency=4 --max-tasks-per-child=25. The --concurrency cap is the single most important lever against OOM kills; --max-tasks-per-child recycles the process periodically so allocator fragmentation cannot inflate RSS over a long run.

Step 3: Configure acks_late with Retry Backoff

acks_late makes redelivery-on-crash possible; exponential backoff with jitter keeps a transient dependency outage (a flaky tile server, a throttled object store) from becoming a thundering herd of synchronized retries.

from celery import Celery

app = Celery("atlas", broker="redis://localhost:6379/0",
             backend="redis://localhost:6379/1")

app.conf.update(
    task_acks_late=True,              # ack only after success → redeliver on crash
    task_reject_on_worker_lost=True,  # requeue if the worker dies mid-task
    worker_prefetch_multiplier=1,     # long tasks: never hoard the queue
    task_serializer="json",
    result_serializer="json",
    result_expires=86400,             # keep results a full day for slow batches
)

Setting worker_prefetch_multiplier=1 is essential for long-running renders: the default (4) lets one worker reserve four messages, so three sheets sit idle behind a slow render instead of flowing to free workers.

Step 4: Route Exhausted Failures to a Dead-Letter Queue

When autoretry_for runs out of attempts, on_failure fires. Publish the sheet spec and traceback to a dedicated dead-letter queue instead of letting the exception vanish.

import traceback
from celery import Task

class RenderTask(Task):
    autoretry_for = (IOError, TimeoutError)
    retry_backoff = 2          # 2s, 4s, 8s, ... exponential
    retry_backoff_max = 300    # cap individual backoff at 5 minutes
    retry_jitter = True        # randomize to avoid synchronized retries
    max_retries = 5

    def on_failure(self, exc, task_id, args, kwargs, einfo):
        sheet = kwargs.get("sheet") or (args[0] if args else {})
        app.send_task(
            "atlas.dead_letter",
            kwargs={
                "sheet": sheet,
                "task_id": task_id,
                "error": repr(exc),
                "traceback": traceback.format_exc(),
            },
            queue="dead_letter",     # a separate queue nothing auto-consumes
        )

A dead_letter handler simply persists the payload (to a table or a bucket) for a human to inspect. Replay is then a matter of re-dispatching those sheet specs once the underlying data is fixed.

Step 5: Chunk the Manifest and Aggregate with a Chord

A 5,000-sheet manifest must not become one 5,000-element group message — that overflows broker frame limits. Split it into bounded chunks, dispatch each chunk’s rows as a group, and fan back in with a chord callback that runs the batch-level preflight and progress update.

from celery import chord, group

def chunked(seq, size):
    for i in range(0, len(seq), size):
        yield seq[i:i + size]

def dispatch_chunk(sheet_ids, atlas_id):
    header = group(render_sheet.s(sid) for sid in sheet_ids)
    callback = finalize_chunk.s(atlas_id=atlas_id)
    return chord(header)(callback)   # returns the callback's AsyncResult

Passing only sheet_id — not the full spec — into the signature keeps each group message tiny; the task loads the row from the shared manifest store. This is what makes a manifest of any size dispatchable.

Complete Working Code Example

"""
atlas_orchestrator.py

Distributed map-sheet rendering with Celery:
  idempotent render task · acks_late + retry/backoff · dead-letter routing ·
  memory-aware concurrency · chunked group/chord dispatch · progress tracking.

Run a worker:
  celery -A atlas_orchestrator worker --concurrency=4 --max-tasks-per-child=25 \
         -Q render,dead_letter --loglevel=INFO
Monitor:
  celery -A atlas_orchestrator flower
"""

import hashlib
import json
import os
import time
import traceback
from pathlib import Path

import psutil
from celery import Celery, Task, chord, group

# ── Broker / backend configuration ──────────────────────────────────────────

app = Celery(
    "atlas",
    broker=os.environ.get("CELERY_BROKER", "redis://localhost:6379/0"),
    backend=os.environ.get("CELERY_BACKEND", "redis://localhost:6379/1"),
)
app.conf.update(
    task_acks_late=True,
    task_reject_on_worker_lost=True,
    worker_prefetch_multiplier=1,
    task_serializer="json",
    result_serializer="json",
    accept_content=["json"],
    result_expires=86400,
    task_routes={
        "atlas.render_sheet": {"queue": "render"},
        "atlas.dead_letter": {"queue": "dead_letter"},
    },
)

MANIFEST_STORE = os.environ.get("MANIFEST_STORE", "/srv/atlas/manifests")
OUTPUT_ROOT = os.environ.get("OUTPUT_ROOT", "/srv/atlas/out")


# ── Idempotency: deterministic output key from a content hash ───────────────

def render_output_key(sheet: dict) -> str:
    fingerprint = {
        "bbox": [round(c, 6) for c in sheet["bbox"]],
        "crs": sheet["crs"],
        "layers": sorted(sheet["layers"]),
        "style_version": sheet["style_version"],
        "dpi": sheet["dpi"],
    }
    blob = json.dumps(fingerprint, sort_keys=True, separators=(",", ":"))
    digest = hashlib.sha256(blob.encode("utf-8")).hexdigest()[:16]
    return f"{sheet['atlas_id']}/{sheet['sheet_id']}-{digest}.tiff"


def load_sheet(atlas_id: str, sheet_id: str) -> dict:
    """Load one sheet spec from the shared manifest — messages carry only IDs."""
    path = Path(MANIFEST_STORE) / f"{atlas_id}.json"
    manifest = json.loads(path.read_text())
    return manifest["sheets"][sheet_id]


def render_to_disk(sheet: dict, out_path: Path) -> None:
    """
    Placeholder for the real rendering call. In production this delegates to the
    Agg/GeoPandas/rasterio export pipeline, writing a print-ready sheet.
    Kept isolated so the orchestration logic stays testable.
    """
    from render_pipeline import render_sheet_to_tiff  # your export module
    render_sheet_to_tiff(sheet, str(out_path))


# ── The render task: idempotent, retrying, dead-lettering ───────────────────

class RenderTask(Task):
    name = "atlas.render_sheet"
    autoretry_for = (IOError, TimeoutError, ConnectionError)
    retry_backoff = 2
    retry_backoff_max = 300
    retry_jitter = True
    max_retries = 5

    def on_failure(self, exc, task_id, args, kwargs, einfo):
        app.send_task(
            "atlas.dead_letter",
            kwargs={
                "ref": kwargs.get("ref") or (args[0] if args else None),
                "task_id": task_id,
                "error": repr(exc),
                "traceback": traceback.format_exc(),
            },
            queue="dead_letter",
        )


@app.task(bind=True, base=RenderTask)
def render_sheet(self, ref: dict) -> dict:
    """
    ref = {"atlas_id": ..., "sheet_id": ...}  — small message, no heavy payload.
    Returns a tiny result dict so the chord join stays lightweight.
    """
    sheet = load_sheet(ref["atlas_id"], ref["sheet_id"])
    key = render_output_key(sheet)
    out_path = Path(OUTPUT_ROOT) / key
    out_path.parent.mkdir(parents=True, exist_ok=True)

    # Idempotency shortcut: identical inputs already rendered → skip the work.
    if out_path.exists() and out_path.stat().st_size > 0:
        return {"sheet_id": ref["sheet_id"], "key": key, "status": "cached"}

    # Write to a temp path then atomically rename, so a crash mid-write never
    # leaves a truncated file that the .exists() check would treat as complete.
    tmp_path = out_path.with_suffix(out_path.suffix + f".{self.request.id}.part")
    render_to_disk(sheet, tmp_path)
    os.replace(tmp_path, out_path)

    return {"sheet_id": ref["sheet_id"], "key": key, "status": "rendered"}


@app.task(name="atlas.dead_letter")
def dead_letter(ref, task_id, error, traceback):  # noqa: A002
    """Persist permanently-failed sheets for inspection and later replay."""
    dlq_path = Path(OUTPUT_ROOT) / "_dead_letter" / f"{task_id}.json"
    dlq_path.parent.mkdir(parents=True, exist_ok=True)
    dlq_path.write_text(json.dumps(
        {"ref": ref, "task_id": task_id, "error": error, "traceback": traceback},
        indent=2,
    ))
    return {"dead_lettered": task_id}


# ── Fan-in: chord callback validates a chunk and reports progress ───────────

@app.task(name="atlas.finalize_chunk")
def finalize_chunk(results, atlas_id: str, chunk_index: int, total_chunks: int):
    """
    Runs once per chunk after every render in the group has stored its result.
    'results' is the list of small dicts returned by render_sheet.
    """
    rendered = [r for r in results if r["status"] in ("rendered", "cached")]
    # Progress is written to the backend where a UI / Flower sidecar can read it.
    app.backend.set(
        f"progress:{atlas_id}",
        json.dumps({
            "chunk": chunk_index + 1,
            "total_chunks": total_chunks,
            "sheets_done": len(rendered),
            "updated": time.time(),
        }),
    )
    return {"atlas_id": atlas_id, "chunk_index": chunk_index,
            "count": len(rendered)}


# ── Dispatcher: chunk the manifest, fan out group/chord per chunk ───────────

def chunked(seq, size):
    for i in range(0, len(seq), size):
        yield seq[i:i + size]


def run_batch(atlas_id: str, chunk_size: int = 200) -> list:
    """
    Split a manifest of arbitrary size into bounded chunks and dispatch each as
    a chord. Returns the list of chord AsyncResults (one per chunk) so a caller
    can await completion or poll progress:{atlas_id}.
    """
    manifest = json.loads((Path(MANIFEST_STORE) / f"{atlas_id}.json").read_text())
    sheet_ids = list(manifest["sheets"].keys())
    chunks = list(chunked(sheet_ids, chunk_size))
    total = len(chunks)

    async_results = []
    for idx, chunk in enumerate(chunks):
        header = group(
            render_sheet.s({"atlas_id": atlas_id, "sheet_id": sid})
            for sid in chunk
        )
        callback = finalize_chunk.s(
            atlas_id=atlas_id, chunk_index=idx, total_chunks=total,
        )
        async_results.append(chord(header)(callback))
    return async_results


if __name__ == "__main__":
    # Dispatch a batch of any size; workers drain it under memory-safe concurrency.
    results = run_batch("national-topo-2026", chunk_size=200)
    print(f"Dispatched {len(results)} chunks; poll progress:national-topo-2026")

Performance Optimization Patterns

1. Prefetch multiplier tuning — O(1) fairness. With long renders, keep worker_prefetch_multiplier=1. The default of 4 lets a worker reserve four messages the instant it connects, so a single slow 600 DPI sheet parks three ready sheets behind it while other workers idle. Multiplier 1 turns the queue into a fair work-stealing pool: a worker takes exactly one task, and the next is dispatched only when it is free.

2. Chunk sizing vs broker message limits. Chunk size trades dispatch overhead against message size and chord-join cost. A group of N tasks serialises into one broker message; too large and you hit RabbitMQ’s frame_max (default 128 KB negotiated) or Redis’s proto-max-bulk-len. Because the messages carry only IDs, a few hundred per chunk stays comfortably small. Chunks of 100–300 are the practical sweet spot: large enough that dispatch overhead is amortised, small enough that a chord’s join set stays cheap and one poisoned chunk fails in isolation.

3. Result backend pressure — keep payloads tiny. Every task result is written to the backend and held until result_expires. If tasks return the rendered image bytes, the backend becomes a second, redundant object store and Redis memory balloons until eviction starts dropping results — which stalls chords. Return only a small dict (sheet_id, key, status); the pixels live in the object store, referenced by key.

4. Worker recycling with max_tasks_per_child. Long-lived render processes accumulate memory that the allocator does not always return to the OS — matplotlib figure caches, GDAL dataset handles, fragmentation. Set --max-tasks-per-child=25 (tune to your peak RSS) so each process is retired and replaced after a bounded number of renders, capping the resident set over a multi-hour batch at the cost of periodic fork overhead.

Common Pitfalls and Debugging

1. Memory oversubscription and OOM kills. Symptom: workers vanish mid-batch with no Python traceback; dmesg shows Out of memory: Killed process (celery). Cause: --concurrency set to the CPU count on a memory-bound workload. Fix: derive concurrency from usable_RAM / peak_render_RSS (Step 2), add --max-tasks-per-child to bound fragmentation, and confirm with Flower that per-worker RSS plateaus rather than climbing.

2. Non-idempotent retries writing duplicate files. Symptom: a batch of 5,000 produces 5,050 output objects; a few sheets appear twice with different timestamps. Cause: acks_late redelivered tasks that had already written output under a run-unique or timestamped filename. Fix: make the output key a pure function of the inputs via the content hash (Step 1) and write through a temp-file-plus-atomic-rename so a re-run overwrites deterministically and never leaves a half-written file.

3. Broker message too large. Symptom: dispatch raises ConnectionResetError or RabbitMQ closes the channel with frame_max exceeded; on Redis, OOM command not allowed. Cause: a single group/chord embedding thousands of full sheet specs. Fix: chunk the manifest (Step 5) and pass only IDs in the message — the task loads the heavy spec from the shared store. Keep chunks in the low hundreds.

4. Lost results from an expired or evicted backend. Symptom: a chord callback never fires even though every render finished; Flower shows all header tasks SUCCESS. Cause: result_expires elapsed before the slowest sheet completed, or the Redis backend evicted result keys under maxmemory pressure, so the chord’s join counter can never reach the header size. Fix: raise result_expires past the worst-case batch wall-clock, keep result payloads tiny so eviction is unlikely, and give the result backend a dedicated Redis instance with noeviction so results are never silently dropped.

5. Priority sheets stuck behind a long batch. Symptom: an urgent single-sheet reprint queued during a 5,000-sheet run does not render for hours. Cause: one FIFO queue, so the priority job sits behind the whole batch. Fix: route urgent work to a separate high-priority queue and run a dedicated worker (or -Q priority,render with the priority queue listed first) so latency-sensitive reprints bypass the bulk backlog. Server-side render services that must stay responsive under batch load lean on this same separation described in Headless Rendering Engines.

Conclusion

Batch orchestration turns map rendering from a script you babysit into a service that drains work reliably under a fixed resource budget. The load-bearing disciplines are all about failure and finitude: derive concurrency from memory rather than cores, make every task idempotent so at-least-once delivery is harmless, retry transient faults with jittered backoff, dead-letter the permanent ones so nothing vanishes, and chunk large manifests so no single broker message or chord join grows unbounded. With those in place a run of five sheets and a run of fifty thousand use the same code path, and the chord gate gives you a single, reliable point to validate and assemble the finished atlas. For a focused walkthrough of the worker-level parallelism that underpins this, continue to Parallelizing Map-Sheet Rendering with Celery Workers.

FAQ

Why do my Celery workers get OOM-killed even though CPU usage is low? Map rendering is memory-bound, not CPU-bound. A single 300 DPI sheet can hold 1–4 GB of raster compositing buffers resident. If you set concurrency to the CPU count (say 8) on a 16 GB box, eight simultaneous renders demand far more than 16 GB and the kernel OOM-killer terminates the worker mid-task. Cap concurrency at usable_RAM / peak_render_RSS and set worker_max_tasks_per_child to recycle processes before allocator fragmentation inflates the resident set.

How do I stop retried tasks from writing duplicate map sheets? Celery guarantees at-least-once delivery, so with acks_late a task can run twice after a worker crash. Make the task idempotent: derive a deterministic output key from a content hash of the rendering inputs (bbox, layers, style version, DPI) and write to that exact path. A re-run overwrites the same object rather than appending a new file, so duplicate execution is harmless and the batch stays convergent.

What causes “message too large” errors when dispatching a big manifest? Serialising 5,000 full sheet specifications into a single group or chord produces a broker message that can exceed RabbitMQ’s frame_max or Redis’s proto-max-bulk-len. Do not embed heavy payloads in the message. Pass a chunk index or a manifest row ID and have the task load the sheet spec from a shared store, and split the manifest into bounded chunks so each group message stays well under the broker limit.

Why does my chord callback never fire on very large batches? A chord callback only runs once every header task has stored its result in the result backend. If result_expires elapses before the slowest sheet finishes, or the backend evicts entries under memory pressure, the chord’s join counter can never reach the header size and the callback stalls forever. Raise result_expires beyond the worst-case batch wall-clock time, keep result payloads tiny, and prefer a dedicated Redis instance for the backend so eviction policy is under your control.

Related

• Parallelizing Map-Sheet Rendering with Celery Workers — worker-level parallelism, concurrency flags, and per-sheet task design in depth
• DPI and Resolution Management — the resolution discipline each render task inherits
• Headless Rendering Engines — the server-side render backends that batch workers drive

Back to Print-Ready Export and Batch Generation Workflows