Parallelizing Map Sheet Rendering with Celery Workers

Parallelizing map sheet rendering with Celery workers means turning an atlas into a queue of small, idempotent render_sheet tasks: derive worker concurrency from available memory rather than CPU count, key each task by a content hash so retries never duplicate work, dispatch sheets in bounded group() chunks to keep back-pressure sane, and gate the whole run with a chord() callback that only fires once every sheet is on disk. This recipe renders a 900-sheet atlas across a worker fleet without OOM kills, and it re-runs cleanly after a partial failure because completed sheets are skipped by hash.

Core Algorithm and Workflow

Map rendering is memory-bound, not CPU-bound. A single 300 DPI sheet allocates large raster buffers, so the naive default — Celery’s prefork concurrency equal to the CPU count — over-subscribes RAM and triggers the OOM killer under load. The workflow below sizes concurrency from measured memory, makes each render safe to repeat, and fans the atlas out in controlled waves.

The recipe has four moving parts, and each solves a specific failure mode that appears at atlas scale:

1. Memory-aware concurrency. Measure the peak resident memory of one representative render, then set --concurrency to available RAM divided by that peak with headroom to spare. This is the single most important knob — it is what keeps the fleet off the OOM killer’s radar.
2. Content-hash idempotency. Hash every pixel-affecting input (bbox, CRS, DPI, page size, style version) into a deterministic sheet_id. The task returns early when an artifact for that hash already exists, so acks_late re-delivery and manual re-runs never duplicate or corrupt a sheet.
3. Chunked group() dispatch. Split the sheet list into fixed-size chunks and enqueue each as its own group. The broker never receives one enormous payload, progress is observable chunk by chunk, and a failure is localized rather than atomic across the whole atlas.
4. Chord-gated preflight. Each chunk’s group is wrapped in a chord(). The callback fires exactly once, inside the workers, after every render in the chunk returns — the natural place to assemble the manifest and gate the downstream atlas merge.

This page is the hands-on companion to the Batch Queue Orchestration overview: where that page surveys queue topologies and broker trade-offs, this one is the concrete render_sheet recipe you deploy.

Production-Ready Python Implementation

Three files below make a complete, runnable system: the Celery app config, the idempotent render_sheet task, and the run_batch() dispatcher. Pin celery==5.4.0, psutil==5.9.8, and your renderer of choice — the example calls a render_to_pdf seam that you back with QGIS, Matplotlib, or a headless rendering engine.

Start with the app configuration. The critical choices are acks_late (so a crashed worker’s task is redelivered), worker_prefetch_multiplier=1 (so a worker never reserves a second heavyweight sheet while busy), and worker_max_tasks_per_child (so per-render buffer leaks are reclaimed by recycling the process):

# celery_app.py
import math
import psutil
from celery import Celery

# Measured peak resident memory of one representative 300 DPI sheet render.
# Profile this once with tracemalloc or /usr/bin/time -v and pin it here.
PER_RENDER_PEAK_BYTES = 2 * 1024 ** 3          # ~2 GB per A1 sheet at 300 DPI
MEMORY_HEADROOM = 0.80                          # never commit more than 80% of RAM


def memory_aware_concurrency() -> int:
    """Concurrency = available RAM * headroom / measured per-render peak.

    Map rendering is memory-bound, so deriving concurrency from CPU count
    (Celery's default) over-subscribes RAM and invites the OOM killer.
    """
    available = psutil.virtual_memory().available
    budget = int(available * MEMORY_HEADROOM)
    return max(1, budget // PER_RENDER_PEAK_BYTES)


app = Celery(
    "atlas",
    broker="redis://localhost:6379/0",
    backend="redis://localhost:6379/1",       # result backend is REQUIRED for chords
)

app.conf.update(
    task_acks_late=True,                        # redeliver on worker crash (needs idempotency)
    task_reject_on_worker_lost=True,
    worker_prefetch_multiplier=1,               # one heavyweight sheet reserved at a time
    worker_max_tasks_per_child=24,              # recycle process to reclaim leaked buffers
    task_track_started=True,
    result_expires=3600,
    task_default_queue="render",
)

# Start workers with, e.g.:
#   celery -A celery_app worker --concurrency=$(python -c \
#     "from celery_app import memory_aware_concurrency as c; print(c())")

Next, the task. The content hash is computed from a canonical JSON encoding of the sheet spec so key order never changes the digest, and the early-return check is what makes acks_late redelivery safe:

# tasks.py
import hashlib
import json
import os
import tempfile
from celery_app import app

ARTIFACT_ROOT = "/srv/atlas/sheets"
STYLE_VERSION = "atlas-style-2026.07"          # bump to force a re-render of every sheet


def sheet_id(spec: dict) -> str:
    """Deterministic content hash of every pixel-affecting input.

    bbox, crs, dpi, and page size fix the geometry; STYLE_VERSION invalidates
    the cache when symbology changes. Canonical JSON keeps the digest stable
    regardless of dict key order.
    """
    payload = {
        "bbox": [round(c, 6) for c in spec["bbox"]],
        "crs": spec["crs"],
        "dpi": spec["dpi"],
        "page": spec["page"],
        "style": STYLE_VERSION,
    }
    canonical = json.dumps(payload, sort_keys=True, separators=(",", ":"))
    return hashlib.sha1(canonical.encode("utf-8")).hexdigest()[:16]


def artifact_path(sid: str) -> str:
    return os.path.join(ARTIFACT_ROOT, f"{sid}.pdf")


@app.task(
    bind=True,
    acks_late=True,
    max_retries=3,
    default_retry_delay=30,
    autoretry_for=(IOError,),
    retry_backoff=True,
)
def render_sheet(self, spec: dict) -> dict:
    """Render one map sheet. Idempotent: skips work if the hashed artifact exists."""
    sid = sheet_id(spec)
    out_path = artifact_path(sid)

    # Idempotency gate: a redelivered or re-run task is a no-op once the sheet exists.
    if os.path.exists(out_path) and os.path.getsize(out_path) > 0:
        return {"sheet_id": sid, "path": out_path, "status": "cached"}

    os.makedirs(ARTIFACT_ROOT, exist_ok=True)

    # Write to a temp file in the same directory, then atomically rename.
    # A crash mid-render leaves no half-written .pdf for the gate to accept.
    fd, tmp_path = tempfile.mkstemp(suffix=".pdf", dir=ARTIFACT_ROOT)
    os.close(fd)
    try:
        render_to_pdf(
            bbox=spec["bbox"],
            crs=spec["crs"],
            dpi=spec["dpi"],
            page=spec["page"],
            out_path=tmp_path,
        )
        os.replace(tmp_path, out_path)          # atomic on POSIX same-filesystem rename
    except Exception:
        if os.path.exists(tmp_path):
            os.unlink(tmp_path)
        raise

    return {"sheet_id": sid, "path": out_path, "status": "rendered"}


def render_to_pdf(bbox, crs, dpi, page, out_path):
    """Seam for your renderer (QGIS layout exporter, Matplotlib, Mapnik, ...).

    Whatever backend you use, it must emit a single print-ready PDF at `dpi`
    for the given `bbox` in `crs` sized to `page`, written to `out_path`.
    """
    raise NotImplementedError("wire up your headless renderer here")

Finally, the dispatcher. run_batch() chunks the atlas, wraps each chunk in a chord whose callback assembles a manifest, and returns immediately without blocking on results:

# dispatch.py
from celery import chord, group
from tasks import render_sheet, sheet_id
from celery_app import app

CHUNK_SIZE = 200                                # bound broker payload + failure blast radius


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


@app.task
def assemble_manifest(results: list, chunk_index: int) -> dict:
    """Chord callback: runs once, after every render in the chunk returns.

    This is the preflight gate — verify the count, record which sheets were
    freshly rendered vs. served from the content-hash cache, then hand off
    to the atlas merge.
    """
    rendered = [r for r in results if r["status"] == "rendered"]
    cached = [r for r in results if r["status"] == "cached"]
    manifest = {
        "chunk": chunk_index,
        "total": len(results),
        "rendered": len(rendered),
        "cached": len(cached),
        "sheet_ids": [r["sheet_id"] for r in results],
    }
    # merge_atlas.delay(manifest["sheet_ids"])  # trigger downstream merge here
    return manifest


def run_batch(sheet_specs: list[dict]) -> list:
    """Dispatch an atlas as bounded, chord-gated chunks. Returns async results."""
    async_results = []
    for idx, chunk in enumerate(chunked(sheet_specs, CHUNK_SIZE)):
        header = group(render_sheet.s(spec) for spec in chunk)
        callback = assemble_manifest.s(chunk_index=idx)
        async_results.append(chord(header)(callback))   # non-blocking dispatch
    return async_results


if __name__ == "__main__":
    # Example: a 3x3 tiled atlas over a metropolitan bbox in EPSG:25832 (UTM 32N).
    minx, miny, maxx, maxy = 545000, 5790000, 560000, 5805000
    step_x, step_y = (maxx - minx) / 3, (maxy - miny) / 3
    specs = [
        {
            "bbox": [minx + i * step_x, miny + j * step_y,
                     minx + (i + 1) * step_x, miny + (j + 1) * step_y],
            "crs": "EPSG:25832",
            "dpi": 300,
            "page": "A1",
        }
        for i in range(3) for j in range(3)
    ]
    for spec in specs:
        print(sheet_id(spec), spec["bbox"])
    run_batch(specs)

Performance Tuning and Cartographic Best Practices

• Profile the per-render peak before you trust the concurrency formula. PER_RENDER_PEAK_BYTES is a guess until you measure it with tracemalloc or /usr/bin/time -v on your largest sheet. A 300 DPI A0 render can hold well over 3 GB; setting the constant too low is exactly how a memory-aware config still gets OOM-killed. Re-measure whenever DPI, page size, or layer count changes.
• Keep worker_prefetch_multiplier=1 for heavyweight tasks. The Celery default prefetches multiple messages per worker process, so a worker mid-render silently reserves the next large sheet and doubles its memory footprint. Setting the multiplier to 1 makes concurrency the only thing that controls how many sheets are in flight per process.
• Bump STYLE_VERSION to invalidate the whole cache in one line. Because symbology is folded into the content hash, a symbology change with an unchanged STYLE_VERSION would serve stale sheets from the idempotency gate. Treat the version string as the cache key for the entire atlas and increment it on every style release.
• Match sheet DPI and page size to the print target, not the screen. The dpi and page fields flow straight into the hash and the renderer, so getting them right up front avoids a full re-render. The DPI trade-offs and QGIS export mechanics are covered in batch exporting 300 DPI PDFs from QGIS with Python and, more broadly, in DPI and Resolution Management.
• Tune CHUNK_SIZE to your broker, not your atlas. Larger chunks amortize chord bookkeeping but enlarge the broker payload and the blast radius of a single failure. For Redis-backed brokers, chunks of 100–300 signatures keep message size comfortable while still letting a failed chunk retry in isolation.
• Give idempotent atomic writes to the artifact, never a partial file. The mkstemp + os.replace pattern guarantees the idempotency gate only ever sees complete sheets, so a worker killed mid-render leaves nothing for a redelivered task to mistake for finished work.

Integration and Next Steps

The chord callback is your integration seam. Its manifest lists every sheet_id in the chunk, which is exactly what a downstream merge step needs to assemble a multi-page atlas PDF or hand sheets to a plotting service. Because completed sheets are content-addressed, re-running run_batch() after a partial failure re-renders only the missing sheets and the callback still fires with the full set — the preflight gate never lets a half-finished atlas proceed to print.

To scale beyond a single machine, point the broker and result backend at shared Redis and run identical worker fleets on each node; each node computes its own memory_aware_concurrency(), so heterogeneous hardware self-balances. When rendering shifts from static PDF sheets to tile pyramids, the same chord-gated dispatch drives a headless rendering engine instead of a PDF exporter — the task contract (spec in, content-hashed artifact out) is unchanged.

Frequently Asked Questions

Why do my Celery workers get OOM-killed when I raise concurrency?

Celery’s default prefork concurrency equals the CPU count, but map rendering is memory-bound: a single 300 DPI A1 sheet can hold 1.5–3 GB of raster buffers at peak. Eight CPU-derived worker processes each rendering a large sheet will exceed physical RAM and the kernel OOM-killer terminates them. Set concurrency from psutil.virtual_memory().available divided by the measured per-render peak, and add worker_max_tasks_per_child so leaked buffers are reclaimed through process recycling.

How does a content hash make the render task idempotent?

Hash every input that affects the pixels — bbox, CRS, DPI, page size, and a style-version string — into a stable sheet_id with hashlib.sha1 over canonical JSON. The task checks for an existing artifact at the path derived from that hash and returns early if it is present. Because acks_late=True re-delivers a task if a worker dies mid-render, this early return is what stops a retry from producing a duplicate or corrupt half-written sheet.

Why use a chord instead of just waiting on the group result?

A chord() runs its callback exactly once, after every task in the header group completes, entirely inside the workers — the dispatching process never blocks on result.get(), which would tie up a web request or pin a coordinator. The callback receives the list of per-sheet return values, so it is the natural place to assemble the manifest, verify sheet counts, and trigger the atlas merge as a preflight gate before print.

Why chunk the group instead of dispatching every sheet at once?

A single group() of tens of thousands of signatures is serialized and pushed to the broker in one payload, spiking broker memory and latency, and the whole chord fails atomically if any one task errors. Fixed-size chunks keep each broker message bounded, let you observe progress chunk by chunk, and localize a failure to one chunk’s chord rather than the entire atlas run.

Related

• Batch Queue Orchestration — the parent overview covering queue topologies, broker choice, and retry semantics that frame this render_sheet recipe.
• Batch Exporting 300 DPI PDFs from QGIS with Python — the per-sheet QGIS export that backs the render_to_pdf seam used inside each Celery task.
• DPI and Resolution Management — the resolution strategy that sets the dpi and page inputs feeding the content hash.

Back to Batch Queue Orchestration