Orchestrating Marine Pipelines with Dask & Prefect

Orchestrating a marine pipeline means turning a sequence of heavy geospatial stages — ingest, clean, grid, publish — into a single reproducible run that survives worker crashes, retries only what failed, and never double-writes a product, and it belongs squarely under Deployment & Cloud Infrastructure for Marine Pipelines. The workload is awkward for a plain script: gridding a multibeam survey needs a Dask cluster of workers holding gigabytes of soundings, cleaning is embarrassingly parallel across survey lines, and publishing is a single careful write to object storage. Prefect supplies the control plane — typed tasks, retries, timeouts, result persistence, and a run history you can audit — while Dask supplies the data plane, spreading array computation across workers that scale up for the gridding stage and back down when it finishes. The failure modes are specific and unforgiving: a worker that exceeds its memory limit is killed mid-graph, a non-idempotent retry writes the same tile twice to a bucket, and a version skew between the scheduler and its workers corrupts the task graph silently. This page shows how to wire the two systems together so a marine pipeline runs the same way every time and fails loudly when it fails at all.

Reference Configuration #

The table below captures the Prefect task controls and Dask cluster parameters that govern a production run. The Prefect rows bound how each stage retries and times out; the Dask rows size the compute cluster and its workers for the gridding stage, which is the memory-heaviest part of the pipeline.

The governing principle: every task must be idempotent and every retry must be safe. Prefect will re-run a task on failure, and Dask will recompute a lost partition; if either action can write a second copy of a product or advance state twice, the pipeline is not correct no matter how well it performs.

Orchestration Architecture #

The diagram shows the Prefect flow driving a Dask cluster: the flow submits the four stages, the Dask scheduler distributes the gridding graph across workers, and only the publish stage writes to the object store, guarded by an idempotent key.

Memory-Constrained Python Implementation #

The flow below is the production artifact. Each stage is a typed @task with retries and a timeout; the flow attaches a DaskTaskRunner so tasks execute on a distributed cluster, and the gridding stage builds its result on Dask arrays so soundings stream through workers block by block rather than loading a full survey into one process. Every task derives a deterministic cache key from its inputs, so a retry recomputes to the same object-store key instead of writing a second copy.

import hashlib
import logging
from dataclasses import dataclass

import dask.array as da
import numpy as np
import xarray as xr
from prefect import flow, task, get_run_logger
from prefect.tasks import task_input_hash
from prefect_dask import DaskTaskRunner
from scipy.spatial import cKDTree

logging.basicConfig(level=logging.INFO)


class StageError(RuntimeError):
    """Raised when a pipeline stage cannot produce a valid result."""


@dataclass(frozen=True)
class SurveyRef:
    """Immutable pointer to a survey's inputs; hashable for idempotent keys."""
    survey_id: str
    source_uri: str
    grid_res_m: float


def _run_key(*parts: object) -> str:
    """Deterministic cache key so retries recompute to the same output key."""
    digest = hashlib.sha256("::".join(map(str, parts)).encode()).hexdigest()
    return digest[:16]


@task(retries=3, retry_delay_seconds=[10, 30, 90], timeout_seconds=600,
      cache_key_fn=task_input_hash, persist_result=True)
def ingest(ref: SurveyRef) -> str:
    """Stage soundings into a partitioned intermediate; return its URI."""
    logger = get_run_logger()
    dst = f"s3://cms-intermediate/{ref.survey_id}/soundings-{_run_key(ref.source_uri)}.parquet"
    logger.info("ingest %s -> %s", ref.source_uri, dst)
    # ... read source_uri, normalize CRS, write partitioned parquet at dst ...
    return dst


@task(retries=3, retry_delay_seconds=[10, 30, 90], timeout_seconds=900,
      cache_key_fn=task_input_hash, persist_result=True)
def clean(soundings_uri: str) -> str:
    """Despike survey lines in parallel; return the cleaned intermediate URI."""
    logger = get_run_logger()
    dst = soundings_uri.replace("soundings-", "clean-")
    logger.info("clean %s -> %s", soundings_uri, dst)
    # ... per-line spike removal over Dask partitions, write to dst ...
    return dst


@task(retries=2, retry_delay_seconds=[30, 120], timeout_seconds=1800,
      cache_key_fn=task_input_hash, persist_result=True)
def grid(clean_uri: str, ref: SurveyRef) -> str:
    """
    Grid cleaned soundings to a raster on Dask arrays. The output grid is
    built lazily block by block so no worker holds the full survey; each
    block queries a KD-tree of the soundings covering its footprint.
    """
    logger = get_run_logger()
    xs, ys, zs = _load_soundings(clean_uri)          # 1-D dask arrays
    tree = cKDTree(np.column_stack([xs.compute(), ys.compute()]))
    x0, x1, y0, y1 = xs.min().compute(), xs.max().compute(), ys.min().compute(), ys.max().compute()
    res = ref.grid_res_m
    gx = np.arange(x0, x1, res)
    gy = np.arange(y0, y1, res)

    def _block(block: np.ndarray, block_info=None) -> np.ndarray:
        loc = block_info[None]["array-location"]
        yy = gy[loc[0][0]:loc[0][1]]
        xx = gx[loc[1][0]:loc[1][1]]
        mesh_x, mesh_y = np.meshgrid(xx, yy)
        dist, idx = tree.query(np.column_stack([mesh_x.ravel(), mesh_y.ravel()]), k=8)
        w = 1.0 / np.maximum(dist, 1e-6) ** 2
        zsel = zs.compute()[idx]
        out = np.sum(w * zsel, axis=1) / np.sum(w, axis=1)
        return out.reshape(mesh_x.shape).astype("float32")

    lazy = da.zeros((gy.size, gx.size), chunks=(512, 512), dtype="float32")
    grid_arr = da.map_blocks(_block, lazy, dtype="float32")
    depth = xr.DataArray(grid_arr, dims=("y", "x"), coords={"y": gy, "x": gx}, name="depth")

    dst = clean_uri.replace("clean-", "grid-").replace(".parquet", ".zarr")
    depth.to_dataset().chunk({"y": 512, "x": 512}).to_zarr(
        dst, mode="w", consolidated=True)
    logger.info("grid -> %s (%d x %d)", dst, gy.size, gx.size)
    return dst


@task(retries=3, retry_delay_seconds=[10, 30, 90], timeout_seconds=300,
      cache_key_fn=task_input_hash, persist_result=True)
def publish(grid_uri: str, ref: SurveyRef) -> str:
    """Write the final COG + Zarr product under an idempotent key."""
    logger = get_run_logger()
    key = _run_key(ref.survey_id, grid_uri)
    dst = f"s3://cms-products/{ref.survey_id}/dem-{key}.tif"
    if _object_exists(dst):
        logger.info("publish skipped, product already at %s", dst)
        return dst  # idempotent: a retry never double-writes
    logger.info("publish %s -> %s", grid_uri, dst)
    # ... encode COG + Zarr, validate, write manifest last ...
    return dst


def _load_soundings(uri: str) -> tuple[da.Array, da.Array, da.Array]:
    raise NotImplementedError  # reads x, y, z dask arrays from uri


def _object_exists(uri: str) -> bool:
    raise NotImplementedError  # HEAD the object; True if already published


@flow(name="marine-dem-pipeline",
      task_runner=DaskTaskRunner(
          cluster_kwargs={"n_workers": 4, "threads_per_worker": 2,
                          "memory_limit": "8GiB"},
          adapt_kwargs={"minimum": 2, "maximum": 16}))
def marine_dem_pipeline(ref: SurveyRef) -> str:
    """End-to-end DEM run: ingest -> clean -> grid -> publish."""
    logger = get_run_logger()
    logger.info("starting pipeline for survey %s", ref.survey_id)
    soundings = ingest(ref)
    cleaned = clean(soundings)
    gridded = grid(cleaned, ref)
    product = publish(gridded, ref)
    logger.info("pipeline complete: %s", product)
    return product

The memory guarantee lives in grid: da.map_blocks builds the output raster one 512×512 tile at a time, and each tile only pulls the soundings its footprint needs. No worker ever holds the whole survey grid. The adapt_kwargs band lets the scheduler add workers when the gridding graph is wide and release them once the pipeline moves to the single-writer publish stage. The choice between Dask and an alternative distributed engine for this gridding workload — including when Ray’s actor model fits better — is examined in Dask vs Ray for Bathymetric Processing at Scale.

Validation Gates and Quality Control #

Three gates make a run safe to retry and auditable after the fact.

1. Idempotent task keys. Every task’s output URI is derived from a deterministic hash of its inputs (_run_key / task_input_hash). Re-running the flow with the same SurveyRef must produce the identical set of keys. Assert this in a smoke test: run the flow twice against a fixture survey and confirm the product URIs match exactly.

2. Result persistence. With persist_result=True, a flow resumed after a mid-pipeline failure reuses completed stage outputs instead of recomputing them. Verify by killing the flow after clean and resuming: ingest and clean must be served from the result store, not re-executed.

3. Run manifest. After publish, the flow writes a manifest recording each stage’s status, output URI, and duration. A run with any stage not in a terminal Completed state must not advertise a product.

from prefect import flow, task


@task
def write_run_manifest(ref: SurveyRef, stage_uris: dict[str, str],
                       product_uri: str) -> str:
    """Persist an auditable record of the run; the contract for replay."""
    import json
    from prefect.runtime import flow_run

    manifest = {
        "survey_id": ref.survey_id,
        "flow_run_id": str(flow_run.get_id()),
        "grid_res_m": ref.grid_res_m,
        "stages": stage_uris,
        "product": product_uri,
        "state": "completed",
    }
    dst = f"s3://cms-products/{ref.survey_id}/run-manifest.json"
    # ... write manifest bytes to dst, last, after product is validated ...
    return json.dumps(manifest)

The pipeline health metric is the effective retry rate — how often stages fail and re-run. Over a rolling window of runs, define it as

$$q = \frac{n_{\text{task attempts}}}{n_{\text{task successes}}}, \qquad \text{investigate when } q > 1.5$$

A ratio near 1 means tasks mostly succeed first try; a sustained ratio above 1.5 means half again as many attempts as successes — a signal of chronic worker loss, an undersized memory limit, or a flaky object store, not healthy transient recovery.

Common Failure Modes and Diagnosis #

Dask Worker OOM and Nanny Kill #

Symptom: The gridding stage stalls, worker logs show Worker exceeded 95% memory budget. Restarting or KilledWorker, and the task retries indefinitely.

Root cause: A block computation or a .compute() pulled more into one worker than its memory_limit allows — often a survey with denser soundings than the sizing assumed, or an accidental full-array .compute() inside a block function.

Diagnosis: Open the Dask dashboard’s worker memory panel during the run; a saw-tooth climbing to the limit before a restart confirms it. Reduce the block/chunk size, raise memory_limit, or move the KD-tree query to stream soundings per block instead of holding zs.compute() resident. The same OOM signature under Kubernetes-managed workers, and how the pod memory request interacts with the nanny limit, is covered in Handling OOM Kills in Bathymetric Kubernetes Jobs.

Task Graph Too Large for the Scheduler #

Symptom: The flow hangs at graph submission before any worker starts; scheduler memory climbs and the dashboard is unresponsive.

Root cause: Building millions of tiny tasks — for example a per-sounding operation instead of a per-block one — overwhelms the scheduler, which must track every task’s state in memory.

Diagnosis: Check the task count reported at submission; if it is in the millions, the graph is too fine. Coarsen the chunking so each task processes a 512×512 block rather than individual points, and persist intermediates to object storage between stages so the graph is bounded per stage rather than spanning the whole pipeline.

Non-Idempotent Retries Double-Writing to Object Storage #

Symptom: Two copies of a product appear under different keys, or a manifest references a URI that a retry overwrote with different bytes.

Root cause: A task computed its output key from a timestamp, a random UUID, or run-scoped state instead of a deterministic hash of its inputs, so a retry wrote to a fresh key.

Diagnosis: Grep the stage for any non-deterministic input to its destination key. Every write target must derive from _run_key(inputs) and the writer must check existence first, as publish does. Re-run the flow twice against a fixture and assert identical product URIs; a mismatch proves the task is not idempotent.

Scheduler–Worker Version Skew #

Symptom: Tasks fail with cryptic deserialization errors (unpickling, KeyError on a graph key) that appear only on distributed execution, never in a local run.

Root cause: The scheduler and workers run different Dask/distributed (or NumPy) versions, so the serialized task graph deserializes incorrectly on the worker.

Diagnosis: Compare dask.__version__ and distributed.__version__ across scheduler and every worker (the dashboard’s info page lists them). Pin the entire stack in one image and deploy scheduler and workers from that identical image — the reproducibility discipline detailed in Containerizing Marine Geospatial Pipelines with Docker.

Pipeline Integration and Downstream Handoff #

The flow is the spine that connects the domain’s other stages. Its ingest and clean tasks wrap the bathymetric processing logic; its grid task produces the array that the terminal publish task hands to the export path in Publishing COG & Zarr to Object Storage, which encodes, validates, and writes the final product. Because the flow owns retries and idempotency, it guarantees the publish stage runs exactly once per successful run — the precondition that keeps the object store free of torn or duplicated products.

How the flow’s workers are actually provisioned depends on the runtime. Under a container orchestrator, each Dask worker is a pod whose memory request and limit must bracket the memory_limit the nanny enforces; Kubernetes Job Scheduling for Bathymetric Processing covers sizing the job, its resource requests, and restart policy so the scheduler and workers come up together. All of them must run the same pinned image so no version skew can corrupt the graph.

The run manifest is the handoff contract. A downstream cataloguer or tile server reads it to learn which product URI a completed run produced and whether every stage reached a terminal Completed state:

{
  "survey_id": "gom-2026-07-13-mb01",
  "flow_run_id": "b3f1c9a2-8e44-4d17-9a0c-7f2e1d6b5a90",
  "grid_res_m": 2.0,
  "stages": {
    "ingest":  {"state": "Completed", "uri": "s3://cms-intermediate/gom-.../soundings-1a2b.parquet"},
    "clean":   {"state": "Completed", "uri": "s3://cms-intermediate/gom-.../clean-1a2b.parquet"},
    "grid":    {"state": "Completed", "uri": "s3://cms-intermediate/gom-.../grid-1a2b.zarr"},
    "publish": {"state": "Completed", "uri": "s3://cms-products/gom-.../dem-9f3c.tif"}
  },
  "product": "s3://cms-products/gom-.../dem-9f3c.tif",
  "state": "completed",
  "written_at": "2026-07-13T09:14:52Z"
}

A consumer treats a run whose state is anything but completed as unpublished and ignores its intermediate URIs. For orchestration internals — task runner configuration, result stores, and deployment scheduling — consult the Prefect documentation and the Dask distributed documentation, and for the array-scaling tradeoff behind the gridding stage see Dask vs Ray for Bathymetric Processing at Scale.

---

Related #

• Kubernetes Job Scheduling for Bathymetric Processing — provisioning Dask workers as pods with correct resource requests
• Containerizing Marine Geospatial Pipelines with Docker — the pinned image that keeps scheduler and workers version-matched
• Publishing COG & Zarr to Object Storage — the export stage the publish task drives
• Dask vs Ray for Bathymetric Processing at Scale — choosing the distributed engine behind the gridding stage

Up: Deployment & Cloud Infrastructure for Marine Pipelines