Article

Running Raster Warps as Celery Tasks

Split the destination into windows, enqueue one task per window carrying only paths and offsets, and let a chord fire a publish task once the group completes. Each window writes to a unique temporary key and is renamed on success, so a redelivered task is harmless. It builds on the Distributed Task Queues for Spatial Jobs guide, part of the broader Spatial Batch Processing & Async Workflows reference.

Prerequisites

  • Python 3.10 or later
  • pip install "celery[redis]>=5.3" rasterio
  • A Redis or RabbitMQ broker reachable from every worker
  • Shared storage both the enqueuing process and the workers can write to

How the Work Is Split

The destination grid, not the source, decides the task boundaries. Deriving windows from the output means each task owns a disjoint region and no two tasks ever write the same pixels.

Plan, fan out, publish A planning step reads the source header and computes the destination transform and window grid. A group of independent warp tasks each handle one window. A chord callback runs once every window has succeeded and publishes the mosaic and its overviews. plan dst transform + grid warp_window 0,0 warp_window 0,512 warp_window 512,0 warp_window 512,512 publish chord callback

The chord is what makes this more than a fan-out. Overviews and a final manifest can only be built once every window exists, and a callback that fires on group completion expresses that without anyone polling.

Complete Working Implementation

# gistools/warp_tasks.py
from __future__ import annotations

import os
from dataclasses import dataclass

import rasterio
from celery import Celery, chord
from rasterio.warp import calculate_default_transform, reproject, Resampling
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, worker_prefetch_multiplier=1,
                task_reject_on_worker_lost=True, task_serializer="json")

TILE = 1024


@app.task(bind=True, max_retries=4, retry_backoff=True, retry_jitter=True,
          autoretry_for=(ConnectionError, TimeoutError))
def warp_window(self, src: str, dst_dir: str, dst_crs: str,
                col_off: int, row_off: int, width: int, height: int) -> str:
    """Warp one destination window and write it as an independent GeoTIFF."""
    with rasterio.open(src) as s:
        transform, full_w, full_h = calculate_default_transform(
            s.crs, dst_crs, s.width, s.height, *s.bounds)
        win = Window(col_off, row_off, width, height)
        win_transform = rasterio.windows.transform(win, transform)

        profile = s.profile | {
            "crs": dst_crs, "transform": win_transform,
            "width": width, "height": height,
            "tiled": True, "blockxsize": 512, "blockysize": 512,
        }
        final = os.path.join(dst_dir, f"part_{row_off}_{col_off}.tif")
        tmp = f"{final}.{self.request.id}.tmp"

        with rasterio.open(tmp, "w", **profile) as d:
            for band in range(1, s.count + 1):
                reproject(
                    source=rasterio.band(s, band),
                    destination=rasterio.band(d, band),
                    src_transform=s.transform, src_crs=s.crs,
                    dst_transform=win_transform, dst_crs=dst_crs,
                    resampling=Resampling.bilinear,
                )
    os.replace(tmp, final)
    return final


@app.task
def publish_mosaic(parts: list[str], dst: str) -> str:
    """Build a VRT over the finished parts and add overviews."""
    from osgeo import gdal
    vrt = gdal.BuildVRT(dst + ".vrt", sorted(parts))
    vrt.BuildOverviews("AVERAGE", [2, 4, 8, 16])
    vrt = None
    return dst + ".vrt"


def submit(src: str, dst_dir: str, dst: str, dst_crs: str):
    with rasterio.open(src) as s:
        _, w, h = calculate_default_transform(
            s.crs, dst_crs, s.width, s.height, *s.bounds)

    windows = [
        warp_window.s(src, dst_dir, dst_crs, c, r,
                      min(TILE, w - c), min(TILE, h - r))
        for r in range(0, h, TILE)
        for c in range(0, w, TILE)
    ]
    return chord(windows)(publish_mosaic.s(dst))

Step Annotations

  1. The destination transform is recomputed in every task rather than passed in. It is derived deterministically from the source header and the target CRS, so every worker arrives at the same answer, and the payload stays free of floating-point values that would not survive JSON round-trips identically.
  2. self.request.id is in the temporary name. Two attempts at the same window get different temporary files, so a redelivered task cannot corrupt the other’s partial write.
  3. autoretry_for lists only transient errors. A RasterioIOError from a missing source will not succeed on the fourth attempt, and retrying it wastes ten minutes before the failure surfaces.
  4. Windows are clamped at the edges with min(TILE, w - c). The remainder column and row are narrower than a full tile, and a task that asks for pixels beyond the destination extent fails on the write rather than the read, which is much harder to diagnose.
  5. The chord returns parts as paths, not as arrays. The callback needs to know which files exist, not what is in them — keeping results small is what stops the backend filling up.

One Named Gotcha: The Chord Callback Never Fires

A chord callback runs only when every task in the group succeeds. One permanently failing window — a corrupt source region, say — leaves the chord pending forever, and because nothing errored loudly the batch appears to be still running.

Two changes make the failure visible:

@app.task(bind=True)
def warp_window(self, ...):
    try:
        ...
    except Exception as exc:
        # Record it, then succeed with a sentinel so the chord can complete.
        record_dead_letter(src, col_off, row_off, exc)
        return ""          # empty path marks a failed window


@app.task
def publish_mosaic(parts: list[str], dst: str) -> str:
    good = [p for p in parts if p]
    if len(good) != len(parts):
        raise RuntimeError(f"{len(parts) - len(good)} window(s) failed; not publishing")
    ...

Turning a failure into a sentinel result rather than an exception lets the chord complete and moves the decision about whether a partial mosaic is acceptable into one place. Some pipelines genuinely want to publish with holes; most do not, and this makes the choice explicit.

The settings that make this safe are few, and each one corresponds to a specific way the naive version loses work.

Four settings and the failure each prevents Without late acknowledgement a killed worker loses its task. Without a prefetch of one, a worker hoards tasks and the pool unbalances. Without reject-on-worker-lost an abrupt death does not requeue. Without a retry ceiling a poison message loops forever. setting without it task_acks_late = True a killed worker silently drops its window worker_prefetch_multiplier = 1 one worker hoards the long windows task_reject_on_worker_lost = True an OOM kill does not requeue the window max_retries = 4 a corrupt source loops until someone notices

Every one of these defaults the other way, which is why a queue that appears to work in development loses tasks the first time a node is reclaimed.

Verification

# Watch the group progress
celery -A gistools inspect active | grep warp_window | wc -l

# Confirm no temporary files survived a successful run
find "$DST_DIR" -name '*.tmp' | wc -l          # expect 0

# Confirm the parts tile the destination without gaps
python - <<'PY'
import glob, rasterio
from shapely.geometry import box
from shapely.ops import unary_union

geoms = []
for p in glob.glob("parts/part_*.tif"):
    with rasterio.open(p) as s:
        geoms.append(box(*s.bounds))
merged = unary_union(geoms)
print("parts:", len(geoms), "pieces after union:", 
      1 if merged.geom_type == "Polygon" else len(merged.geoms))
PY

A union that produces more than one piece means the grid has a gap — usually an off-by-one in the edge clamping. Checking it once, on a small input, is far cheaper than discovering it in a published mosaic.

Choosing a Window Size for Distributed Warps

Window size does more here than it does in a process pool, because each window is also a message, a temporary object and a unit of retry.

Four things window size decides at once Small windows mean many messages and high broker overhead but cheap retries and quick shutdown. Large windows amortise the broker cost but make a retry expensive and a graceful shutdown slow. broker overhead one message, one ack and one result per window favours larger windows cost of a retry a failed window is redone in full favours smaller windows memory per worker the window's arrays must fit alongside the cache favours smaller windows graceful shutdown a worker finishes its current window before stopping favours smaller windows

Three of the four point the same way, which is why a distributed warp usually wants smaller windows than the same job would use in a single process — around a thousand pixels square rather than the whole strip.

Sizing the Window for a Distributed Warp

Window size decides four things at once here, which is more than it does in a single process: the message count, the cost of a retry, the memory each worker needs, and how long a graceful shutdown takes. Three of the four favour smaller windows, and only broker overhead favours larger ones.

The practical consequence is that a distributed warp usually wants windows around a thousand pixels square rather than the whole strip a single-process job might use. At that size a retry costs seconds, a worker asked to stop finishes within a second or two, and the per-message overhead is still a small fraction of the work.

The remainder windows at the right and bottom edges are narrower than a full tile, which is fine for correctness and worth knowing for scheduling: they finish sooner, so a batch that submits them last ends with a burst of quick completions rather than a slow trickle.

Handling Partial Success

A batch of ten thousand windows will sometimes finish with nine thousand nine hundred, and what happens next should be a decision rather than an accident.

Three policies are defensible and they suit different products. Refusing to publish anything unless every window succeeded is right for a product whose consumers assume completeness — a tile set serving a map, say, where a hole is a visible bug. Publishing with holes and recording them is right for an analysis input where the consumer already handles nodata and a delayed delivery is worse than an incomplete one. Publishing to a staging prefix and requiring a human decision is right when the answer genuinely depends on which windows failed.

Whichever you choose, express it in the callback rather than leaving it implicit. The version in the main implementation raises when any window failed, which is the first policy; changing it to publish with a recorded gap list is a two-line change, and having the policy in one visible place is what makes the change reviewable.

The recording matters as much as the decision. A published mosaic with three missing windows and no record of which ones is a product nobody can reason about later; the same mosaic with a gap list in its manifest is one a consumer can work with.

One further detail: give the chord a timeout. A group whose callback waits indefinitely for a task that will never complete is indistinguishable from a batch still running, and a timeout converts that into a failure someone can see.