Orchestrating Multi-Flight Jobs with Celery and Redis

TL;DR: Declare a Celery app on a Redis broker, define an idempotent per-tile task with autoretry_for and exponential backoff, fan out one task per flight tile with a group, join the per-tile results with a chord, and verify every result against the job manifest before marking a field complete.

Why This Task Arises

The moment a farm operation runs more than a handful of flights a week, sequential processing stops finishing inside the window between when imagery lands and when an agronomist needs the maps. A single Python process stitching and index-computing four hundred fields serially takes hours, and one slow upload or one corrupt tile stalls everything behind it. A task queue turns that serial backlog into a fan-out: each flight’s tiles become independent tasks that run across as many workers as you can afford, and a failure isolates to one tile instead of killing the run.

Without disciplined orchestration, though, the queue creates its own failure modes. A task redelivered after a worker crash writes a second, truncated copy of a tile over a good one. A fixed-interval retry against a throttling storage backend turns a brief outage into a sustained retry storm. A run that “finishes” with silently dropped tiles produces a vegetation-index mosaic with holes, and a hole in an NDVI raster reads downstream as bare or failed ground — an error that flows straight into a variable-rate prescription. This guide builds the Celery layer that prevents all three. It is the coarse-grained task distribution described in batch and async orthomosaic processing; the fine-grained per-tile compute is covered separately in parallel tile processing with Dask.

Task flow from scheduler to worker pool A scheduler enqueues one task per flight into two named Redis queues, one for index computation and one for export. A pool of four workers consumes from those queues with late acknowledgement, so a worker lost mid-task returns its message to the queue after the visibility timeout. Completed results are written to a result backend. Scheduler one task per flight, keyed deterministically Redis broker queue: indices long tasks, prefetch 1 queue: export short tasks, kept separate visibility timeout > slowest task Worker pool worker 1 worker 2 worker 3 worker 4 Result backend state per task, expiring after a day Late acknowledgement returns a lost task to its queue; a deterministic task key keeps the redelivery from producing a second output.

Prerequisites

Only the queue packages differ from the parent guide:

TEXT 
celery==5.3.6
redis==5.0.4
rasterio==1.3.10
numpy==1.26.4
BASH 
pip install celery==5.3.6 redis==5.0.4 rasterio==1.3.10 numpy==1.26.4

Input requirements:

  • A reachable Redis instance for both broker and result backend (redis://localhost:6379).
  • Per-tile COG inputs on object storage, each with an explicit CRS — the task carries the expected EPSG (e.g. EPSG:32615) and refuses a mismatch.
  • A job manifest (a list of tile dicts with src_uri, dst_uri, epsg, tile_id) as produced in the parent guide.

Step-by-Step

One task's life, and where it can be lost Four stages of a Celery task: enqueue with a deterministic key derived from the flight, reserve it with a prefetch of one so a worker does not hoard long tasks, process the flight, and acknowledge only after success so a lost worker returns the message to the queue. Enqueue deterministic task key Reserve prefetch 1 for long tasks Process one flight, minutes to hours Acknowledge only on success A prefetch above one on hour-long tasks means a dying worker takes several unstarted flights down with it.

Step 1 — Configure the app and broker

Redis is the broker and the result backend. The critical settings are late acknowledgement (so a task is only removed from the queue after it succeeds), a prefetch multiplier of 1 (so one worker cannot hoard slow tasks), and a visibility timeout longer than the slowest task (so nothing is redelivered while still running).

PYTHON 
# tasks.py
from celery import Celery

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

app.conf.update(
    task_acks_late=True,
    task_reject_on_worker_lost=True,
    worker_prefetch_multiplier=1,
    result_expires=24 * 3600,
    broker_transport_options={"visibility_timeout": 3 * 3600},
    task_serializer="json",
    result_serializer="json",
    accept_content=["json"],
)

Step 2 — Define the idempotent per-tile task

The task computes NDVI for one tile. autoretry_for with retry_backoff gives exponential backoff with jitter for transient I/O errors; the idempotency guard makes redelivery safe.

PYTHON 
# tasks.py (continued)
import os
import numpy as np
import rasterio
from rasterio.windows import Window

RED_BAND, NIR_BAND = 3, 5  # MicaSense RedEdge-MX


def _compute_ndvi(src_path, dst_path, epsg, chunk=1024):
    with rasterio.open(src_path) as src:
        assert src.crs is not None and src.crs.to_epsg() == epsg, (
            f"Expected EPSG:{epsg}, got {src.crs}"
        )
        profile = src.profile.copy()
        profile.update(count=1, dtype="float32", nodata=np.nan,
                       tiled=True, blockxsize=256, blockysize=256, compress="deflate")
        with rasterio.open(dst_path, "w", **profile) as dst:
            for row in range(0, src.height, chunk):
                for col in range(0, src.width, chunk):
                    win = Window(col, row,
                                 min(chunk, src.width - col),
                                 min(chunk, src.height - row))
                    red = src.read(RED_BAND, window=win).astype("float32")
                    nir = src.read(NIR_BAND, window=win).astype("float32")
                    with np.errstate(invalid="ignore", divide="ignore"):
                        ndvi = (nir - red) / (nir + red)
                    dst.write(ndvi.astype("float32"), 1, window=win)


def _validate(path, epsg):
    with rasterio.open(path) as src:
        return src.count == 1 and src.crs.to_epsg() == epsg


@app.task(
    bind=True, name="flights.ndvi_tile",
    autoretry_for=(IOError, OSError, rasterio.errors.RasterioIOError),
    retry_backoff=True, retry_backoff_max=600, retry_jitter=True,
    max_retries=5, acks_late=True,
)
def ndvi_tile(self, tile: dict):
    dst = tile["dst_uri"]
    tmp = f"{dst}.{self.request.id}.tmp"

    # Idempotency: a previous delivery already produced a valid output
    if os.path.exists(dst) and _validate(dst, tile["epsg"]):
        return {"tile_id": tile["tile_id"], "state": "done", "cached": True}

    _compute_ndvi(tile["src_uri"], tmp, tile["epsg"])
    if not _validate(tmp, tile["epsg"]):
        os.remove(tmp)
        raise self.retry(countdown=30)

    os.replace(tmp, dst)  # atomic on the same filesystem / object store
    return {"tile_id": tile["tile_id"], "state": "done", "cached": False}

Step 3 — Fan out per flight and aggregate with a chord

A group runs one ndvi_tile task per tile in parallel; a chord attaches a callback that fires only once every tile in the group has returned, giving a clean per-field join point. Note that task signatures carry only small dicts — never pixel arrays.

PYTHON 
# tasks.py (continued)
from celery import group, chord


@app.task(name="flights.summarise_flight")
def summarise_flight(results, flight_id):
    done = [r for r in results if r["state"] == "done"]
    return {
        "flight_id": flight_id,
        "tiles_total": len(results),
        "tiles_done": len(done),
        "complete": len(done) == len(results),
    }


def dispatch_flight(flight_id: str, tiles: list):
    """Fan out one task per tile, aggregate into a per-flight summary."""
    for t in tiles:
        assert len(str(t)) < 4096, "Task arg too large — pass URIs, not arrays"
    header = group(ndvi_tile.s(t) for t in tiles)
    return chord(header)(summarise_flight.s(flight_id=flight_id))

Step 4 — Driver and result verification

The driver enqueues each flight, then blocks on the chord result to confirm completion. In production the .get() would be replaced by a manifest poll, but for a runnable example it verifies the fan-out end to end.

PYTHON 
# run_batch.py
from tasks import dispatch_flight

FLIGHTS = {
    "F001": [
        {"tile_id": "t0001", "src_uri": "raw/F001/t0001.tif",
         "dst_uri": "ndvi/F001/t0001.tif", "epsg": 32615},
        {"tile_id": "t0002", "src_uri": "raw/F001/t0002.tif",
         "dst_uri": "ndvi/F001/t0002.tif", "epsg": 32615},
    ],
}

if __name__ == "__main__":
    for flight_id, tiles in FLIGHTS.items():
        async_result = dispatch_flight(flight_id, tiles)
        summary = async_result.get(timeout=1800)   # waits for the chord callback
        print(summary)
        assert summary["complete"], (
            f"Flight {flight_id}: {summary['tiles_done']}/{summary['tiles_total']} "
            "tiles done — investigate failed tasks before compositing"
        )
    print("All flights verified complete")

Start the worker and run the driver (bash):

BASH 
celery -A tasks worker --pool prefork --concurrency 8 -Q celery -n w1@%h &
python run_batch.py

Gotchas & Edge Cases

  • Visibility timeout shorter than a task. Redis has no true acknowledgement; Celery hides a delivered task for visibility_timeout and redelivers if no result lands. A 40-minute stitch under a 30-minute timeout runs twice concurrently — wasted compute and a duplicate write. Set the timeout above your slowest task including retries.
Four defaults that are wrong for hour-long tasks A table of four Celery and Redis settings whose defaults suit short tasks and fail on hour-long imagery jobs: a visibility timeout shorter than the task, a high prefetch count, unexpiring results filling Redis, and routing every kind of work through one queue. Setting Wrong value What goes wrong Visibility timeout shorter than the task Default 1 h, tasks run 90 min The task is redelivered while still running Prefetch high on long tasks Default 4 A dying worker strands its reservations Result backend expiry never No expiry set Redis fills with task results Queue routing one queue for everything Default queue Short exports wait behind long mosaics

  • Non-idempotent tasks + late acks. With acks_late=True, a worker killed after finishing but before acknowledging triggers redelivery. Without the temporary-write-then-atomic-rename guard in Step 2, the redelivery overwrites a good tile mid-read by a downstream job. Idempotency is not optional once late acks are on.

  • The large-payload anti-pattern. Passing a NumPy array or raw GeoTIFF bytes as a task argument serialises megabytes through Redis on every dispatch, throttling the broker. The assert len(str(t)) < 4096 guard in Step 3 catches an accidental array argument before it ships.

  • A chord that never fires. If any task in the group raises an unhandled exception that exhausts its retries, the chord callback may hang waiting for a result that never comes. Give tasks a bounded max_retries and let them return a {"state": "failed"} record instead of raising, so the callback always completes and the failure surfaces in the summary.

Parent Guide

This guide is part of Batch and Async Orthomosaic Processing — see there for the full architecture, the job manifest, asyncio object-storage reads, and how this queue layer fits the wider pipeline.

Parent section Batch and Async Orthomosaic Processing Related Parallel Tile Processing with Dask