Distributed Task Queues for Spatial Jobs

A process pool stops being enough the moment a batch outgrows one machine, needs to survive a node restart, or has to share capacity with other work. A task queue buys all three — and introduces a set of failure modes that a pool never had. This guide covers what a spatial task should contain, how to size workers around GDAL rather than around cores, and how to keep results correct when the same task runs twice. It is part of the Spatial Batch Processing & Async Workflows guide.

Prerequisites

• Python 3.10 or later, and a broker — Redis or RabbitMQ for Celery, Redis alone for RQ.
• pip install celery redis rasterio pyogrio, or the equivalent for whichever queue you choose.
• A batch that already works on one machine. Moving a broken pipeline onto a queue makes it harder to debug, not easier — get multiprocessing right first.
• Somewhere shared to read from and write to. A queue distributes compute, not storage, so every worker needs the same view of the data — see Cloud Storage I/O for Spatial Batches.

Problem framing

The seductive thing about a task queue is that the migration looks trivial: replace pool.map with task.delay and you are distributed. What actually changes is the failure model. A process pool lives and dies with its parent, so a crash loses everything and you rerun. A queue outlives any individual worker, which is the point — and it means a task can be delivered twice, a worker can die holding a task, and a result can arrive after the code that asked for it has gone away.

None of that is exotic. All of it produces the same symptom in a geospatial pipeline: an output file written twice, once by a worker that was about to be killed and once by its replacement, with the second write starting before the first finished.

What belongs in a task payload

The single most consequential design decision is what a task carries. Get it wrong and every other problem gets harder.

The replayability point is the one that matters most in practice. A reference payload can be re-enqueued from a dead-letter store six months later and will produce the same result, which is what makes the recovery patterns in Error Handling in Spatial Pipelines work at all.

Step-by-step implementation

1. Define the task around one unit of output

# gistools/tasks.py
from __future__ import annotations

from dataclasses import asdict, dataclass

import rasterio
from celery import Celery
from rasterio.windows import Window

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

app.conf.update(
    task_acks_late=True,            # only ack after the task finishes
    worker_prefetch_multiplier=1,   # do not hoard tasks
    task_reject_on_worker_lost=True,
    task_serializer="json",
    result_expires=3600,
)


@dataclass(frozen=True)
class WarpUnit:
    src: str
    dst: str
    dst_crs: str
    col_off: int
    row_off: int
    width: int
    height: int

    def key(self) -> str:
        return f"{self.dst}:{self.col_off},{self.row_off}"


@app.task(bind=True, max_retries=4, autoretry_for=(OSError,),
          retry_backoff=True, retry_jitter=True)
def warp_window(self, unit: dict) -> dict:
    u = WarpUnit(**unit)
    window = Window(u.col_off, u.row_off, u.width, u.height)

    with rasterio.open(u.src) as src:
        data = src.read(window=window)
        profile = src.profile

    tmp = f"{u.dst}.{self.request.id}.tmp"
    profile.update(width=u.width, height=u.height, crs=u.dst_crs,
                   transform=src.window_transform(window))
    with rasterio.open(tmp, "w", **profile) as dst:
        dst.write(data)

    _publish(tmp, u.dst)             # atomic rename or object-store copy
    return {"key": u.key(), "bytes": data.nbytes}

Three settings in that configuration block do most of the work. task_acks_late means a task is only removed from the queue once it has finished, so a worker killed mid-task returns the work to the queue rather than losing it. worker_prefetch_multiplier=1 stops a worker reserving a dozen tasks it will not get to, which is what makes a mixed-duration batch balance itself. And task_reject_on_worker_lost makes the requeue happen on an abrupt death rather than only on a clean shutdown.

2. Make the task idempotent, because it will run twice

task_acks_late guarantees at-least-once delivery, not exactly-once. That guarantee is the right one — the alternative loses work — but it means the task body has to tolerate being run again.

def _publish(tmp: str, dst: str) -> None:
    """Move a completed temporary file onto its final name, atomically."""
    import os
    os.replace(tmp, dst)     # same filesystem: atomic; no partial dst ever exists

The temporary name includes the task id, so two concurrent attempts write to different temporary files and the rename picks a winner. Both attempts produce identical bytes, so which one wins does not matter — which is the definition of idempotent for this shape of work.

3. Size workers for GDAL, not for cores

The default advice — one worker per core — is wrong for geospatial work, usually by a factor of two or more, because each worker carries a GDAL block cache and a working set that cores do not account for.

The right number is the memory budget divided by the per-worker footprint, capped at core count. Deriving it at start-up from an environment variable — the container limit is available at /sys/fs/cgroup/memory.max on modern kernels — beats pinning it in a config file that nobody revisits when the node size changes.

4. Route heavy and light work to separate queues

A single queue means a thousand cheap metadata tasks sit behind one four-hour mosaic. Two queues, with workers dedicated to each, keeps both moving.

app.conf.task_routes = {
    "gistools.tasks.warp_window": {"queue": "heavy"},
    "gistools.tasks.read_metadata": {"queue": "light"},
    "gistools.tasks.publish_manifest": {"queue": "light"},
}

The split is by resource profile rather than by importance. Heavy tasks are memory-bound and want few workers with large limits; light tasks are I/O-bound and want many workers with small ones. That is the same reasoning the section applies to async I/O, one level up.

Configuration & state management

Workers are separate processes on separate machines, so every setting the task depends on has to reach them explicitly. Three categories behave differently.

Values that are part of the work — the target CRS, the resampling method, the compression — belong in the payload. Putting them in a config file read by the worker means a task enqueued today and run tomorrow can produce different output from the same input, which destroys reproducibility.

Values that are part of the environment — PROJ_DATA, GDAL_CACHEMAX, credentials — belong in the worker’s environment, deployed with the worker. They are properties of the machine, not of the task, and putting them in a payload means a queue full of stale credentials.

Values that tune behaviour — retry limits, timeouts — belong in the queue configuration, where they can be changed without redeploying either. The guidance in Environment Variable Sync applies to the second category unchanged.

Error handling & gotchas

A task that outlives its visibility timeout is delivered again while still running. Two workers then process the same window concurrently. The temporary-file pattern above makes that harmless, but only if the temporary name is unique per attempt — a fixed .tmp suffix reintroduces the race.

Results in the backend are not free. Celery stores every return value by default, and a batch of a hundred thousand tasks fills Redis with results nobody reads. Set result_expires, or turn results off entirely with ignore_result=True for tasks whose output is a file.

A large payload can exceed the broker’s message limit. RabbitMQ’s default frame size will reject a task carrying an embedded geometry over a certain size, and the failure is reported as a connection error rather than as your message is too big. Reference payloads avoid this entirely.

Late acknowledgement plus a long task means a slow shutdown. A worker asked to stop finishes its current task first, which for a four-hour mosaic means four hours. Splitting the mosaic into windowed tasks bounds the shutdown time to one window, which is the practical reason to keep tasks small even when the work would parallelise fine as one unit.

Retries reset the payload, not the side effects. An autoretry_for on a task that has already written half its output will re-run from the top with the half-written file still present. The temporary-then-rename pattern handles it; writing directly to the destination does not.

The failure modes a queue adds are worth listing plainly, because none of them exists in a process pool and each has a specific mitigation.

The third row is the one that turns a bad input into an outage. Without a retry ceiling, a message that segfaults the worker is redelivered forever, and every worker in the pool dies on it in turn.

Verification

# 1. Tasks are small — inspect a queued message.
redis-cli -n 0 LRANGE heavy 0 0 | head -c 400

# 2. Late acknowledgement is really on: kill a worker mid-task and confirm requeue.
celery -A gistools inspect active
kill -9 <worker-pid>
celery -A gistools inspect reserved     # the unit reappears

# 3. Idempotency: run the same unit twice and compare.
python -c "from gistools.tasks import warp_window; \
           u=dict(src='in.tif', dst='out.tif', dst_crs='EPSG:3857', \
                  col_off=0,row_off=0,width=256,height=256); \
           warp_window(u); warp_window(u)"

The second check is the one worth doing before trusting the setup. A configuration that looks right and silently loses work on a kill is indistinguishable from a correct one until the day a node is reclaimed mid-batch.

Performance notes

Broker round-trip time sets the floor on useful task size. With a task taking fifty milliseconds and a round trip of five, a tenth of the throughput goes to coordination. As a rule of thumb, keep tasks above a second of work; below that, batch several units into one task and iterate inside it.

Prefetch is the other lever, and it points the opposite way. A prefetch of one is right for long, variable tasks and wasteful for short, uniform ones, where the worker spends its time waiting for the next message. Uniform sub-second tasks want a prefetch of four or more; the mixed-duration geospatial case almost never does.

Result serialisation deserves a look too. JSON is the safe default and cannot carry a numpy array — which is a feature, since it forces the reference-payload discipline above. If profiling shows serialisation cost, the answer is smaller results, not a faster serialiser.

Operating a queue you did not build

Most of the day-to-day cost of a task queue is not in writing the tasks; it is in the questions that come up once it is running. Four of them account for almost all of it, and each has an answer worth deciding before the first incident rather than during it.

How do I know it is working? Queue depth and completion rate together. Depth alone is ambiguous — a growing queue is fine if the rate is high and the producer is ahead — but depth rising while the rate falls is the signature of a stuck pool. Both are one command away in every broker, and putting them on a dashboard costs an afternoon.

How do I stop it? Purge first, then revoke, in that order. Revoking while the queue still holds work means the workers pull more as you go. A batch identifier in the task headers lets you purge selectively, which matters when several batches share a queue.

How do I re-run the failures? From the dead-letter store, not from the original input. The whole point of a reference payload is that a recorded failure can be re-enqueued directly, and rebuilding the task from the source risks a different payload than the one that failed.

How do I roll out a change safely? Version the task name rather than the task body. A worker running old code and a queue holding new payloads is the failure mode that produces the strangest symptoms, and giving the new shape a new task name means old and new coexist until the queue drains.

Where the queue should not go

It is worth naming the cases where a queue is the wrong answer, because the pattern is attractive enough to be over-applied.

A batch that runs once a night on one machine, finishes in an hour, and has never been interrupted does not need one. A process pool with a checkpoint file gives the same recovery with none of the operational surface.

A pipeline whose stages must run in a strict order with data flowing between them is a workflow rather than a set of tasks. Expressing it as a chain of queue messages works and is harder to reason about than a workflow engine or, frequently, a single script.

And an interactive request — a user waiting for an answer — is a poor fit for a queue whose whole design assumes nobody is waiting. The latency floor of a broker round trip plus a worker pick-up is usually acceptable; the failure mode when the pool is saturated is not, because the user sees a request that never returns rather than an error.

FAQ

One more operational note worth writing down before it is needed: decide, in advance, what happens to in-flight work during a deployment. Rolling workers while a batch is running means some units run under the old code and some under the new, which is fine when the change is a bug fix and produces an inconsistent output set when it is a behaviour change. Draining the queue before deploying, or versioning the task name so old and new coexist, are the two arrangements that make the answer predictable rather than incidental.

Related

• Multiprocessing Geospatial Tasks — the single-machine pattern this replaces.
• Cloud Storage I/O for Spatial Batches — the shared storage a distributed batch depends on.
• Error Handling in Spatial Pipelines — what to do with the failures a queue surfaces.