Conditional Branching in Geospatial DAGs

In short: branch on a value the pipeline computed and recorded, not on an ambient condition read inside a task. A spatial DAG’s branches are almost always driven by properties of the data — how big the extent is, whether the source changed, how many features arrived, whether the geometry is valid — and each of those decisions should be visible in the run history as its own artefact, so that “why did it take the small-extent path last Tuesday?” has an answer.

The alternative, which most pipelines start with, is an if inside a task. It works and it is invisible: the run shows one task that took four minutes, with no record of which of its two very different behaviours it exhibited. Six weeks later, when the output looks wrong, there is no way to tell whether the fast path ran when the slow path should have. Making the branch a first-class part of the graph costs a few lines and buys an answer.

Spatial pipelines branch more than most, and it is worth being clear about why, because it shapes what a good branch looks like here. The inputs vary enormously along axes that matter for execution: a scene may be four megapixels or four hundred, a delivery may contain one municipality or forty, a source may be a tiled COG on the same continent or a striped GeoTIFF behind a slow gateway. A single code path that handles all of those either does the expensive thing to everything, or does the cheap thing and falls over on the tail. Branching is how a pipeline stays honest about a workload whose shape is genuinely bimodal — and the discipline below exists because an unrecorded branch turns that honesty into a mystery.

Prerequisites & Architecture Baseline#

• An orchestrator whose graph can be shaped at runtime: Prefect 2.14+ (plain Python control flow) or Dagster 1.5+ (@graph with conditional outputs or dynamic outputs)
• A cheap way to compute each branch predicate — headers and metadata, not a full read of the data
• Somewhere to record the decision: a task return value, a run tag, or a row in a decisions table
• rasterio ≥ 1.3 and fiona ≥ 1.9 for header-only inspection
• A default branch that is safe rather than fast, for the case where the predicate cannot be computed
• Metrics per branch, so the distribution of decisions is visible over time

Core Principles#

1. Compute the predicate in its own task. A task whose only job is to answer “is this extent larger than the threshold?” produces a value that appears in the run history, can be asserted on in tests, and can be looked up months later. An if buried in a five-hundred-line task produces nothing.

2. Predicates must be cheap. Deciding whether to take the tiled path should not require reading the raster. rasterio.open() reads headers; fiona.open() reads a layer definition; an object store’s HEAD returns size and modification time. If computing the predicate costs as much as the branch it selects, restructure it.

3. Branch on data properties, not on the calendar. “Run the full rebuild on Sundays” encodes a guess about when sources change. “Run the full rebuild when the source digest differs from the last successful run” encodes what actually matters, and it degrades gracefully when the source publishes early.

4. Every branch must terminate in the same contract. Whatever path the data takes, the downstream tasks expect the same format, CRS and guarantees. A fast path that skips validation and a slow path that includes it produce outputs that are not interchangeable, and the difference will be discovered downstream by something that assumed they were.

5. The default branch is the safe one. When the predicate cannot be computed — the header is unreadable, the metadata is missing — take the expensive, thorough path. A pipeline that defaults to “assume nothing changed” when it cannot tell will eventually skip a real update, and that failure is silent.

6. Skipping is a branch too, and it deserves the same treatment. A tile that is up to date should be skipped explicitly, with a recorded reason, rather than dropped by a filter three steps earlier. That is what makes “why is this tile missing?” answerable — see skipping tiles with no new source data.

There is a second reason to lift the decision out of the task, less obvious than auditability but often more valuable: it makes the decision testable without any data. inspect() takes a URI, a previous digest and a current digest, and returns a Decision. Every interesting case — unchanged source, oversized scene, unreadable header — is a three-line test with no fixtures and no I/O. When the same logic lives inside a warp task, testing the “we should have tiled this” case means constructing a 400-megapixel raster, and in practice nobody does, so the branch is never tested at all.

Production Implementation#

The flow below inspects cheaply, records a decision object, and selects a path. Prefect is used here because plain Python control flow makes the shape obvious; the Dagster equivalent uses conditional outputs and reads almost identically.

from __future__ import annotations

from dataclasses import dataclass
from enum import Enum
from pathlib import Path
from typing import Optional

import rasterio
from prefect import flow, task, get_run_logger


class Path_(str, Enum):
    SKIP = "skip"
    SINGLE_PASS = "single_pass"
    TILED = "tiled"


@dataclass(frozen=True)
class Decision:
    """The branch, plus everything needed to justify it later."""

    path: Path_
    reason: str
    megapixels: float
    source_digest: str
    previous_digest: Optional[str]


# Above this the warp will not fit comfortably in one worker's memory, so the
# tiled path is not an optimisation but a correctness requirement.
TILED_THRESHOLD_MP = 400.0


@task
def inspect(source_uri: str, previous_digest: Optional[str], digest: str) -> Decision:
    """Header-only inspection. Never reads pixels."""
    logger = get_run_logger()
    try:
        with rasterio.open(source_uri) as src:
            megapixels = (src.width * src.height * src.count) / 1_000_000
    except rasterio.RasterioIOError as exc:
        # The safe default is the thorough path, never the cheap one.
        logger.warning("could not inspect %s (%s) — taking the tiled path", source_uri, exc)
        return Decision(Path_.TILED, f"inspection failed: {exc}", 0.0, digest, previous_digest)

    if previous_digest is not None and previous_digest == digest:
        return Decision(Path_.SKIP, "source digest unchanged since the last successful run",
                        megapixels, digest, previous_digest)

    if megapixels > TILED_THRESHOLD_MP:
        return Decision(Path_.TILED, f"{megapixels:.0f} Mpx exceeds the {TILED_THRESHOLD_MP:.0f} Mpx limit",
                        megapixels, digest, previous_digest)

    return Decision(Path_.SINGLE_PASS, f"{megapixels:.0f} Mpx fits in one pass",
                    megapixels, digest, previous_digest)


@task
def record_decision(conn, scene_id: str, decision: Decision) -> None:
    """The decision outlives the run. Six weeks from now this is the answer."""
    with conn.cursor() as cur:
        cur.execute(
            "INSERT INTO branch_decisions (scene_id, path, reason, megapixels, "
            "source_digest, decided_at) VALUES (%s, %s, %s, %s, %s, now())",
            (scene_id, decision.path.value, decision.reason,
             decision.megapixels, decision.source_digest),
        )


@flow
def process_scene(scene_id: str, source_uri: str, conn, scratch: Path) -> Optional[Path]:
    digest = source_digest(source_uri)
    decision = inspect(source_uri, previous_digest=last_digest(conn, scene_id), digest=digest)
    record_decision(conn, scene_id, decision)

    if decision.path is Path_.SKIP:
        return None
    if decision.path is Path_.SINGLE_PASS:
        return warp_single_pass(source_uri, scratch)
    return merge_tiles(warp_tiled(source_uri, scratch))

Step-by-Step Walkthrough#

1. Make the decision a value, not a control-flow accident. Decision carries the path and the reason and the measurements behind it. That is what turns a branch into something you can audit.
2. Inspect through headers only. rasterio.open() on a remote COG reads a few kilobytes. The whole predicate costs less than the logging around it, which is what makes it acceptable to run for every scene.
3. Order the conditions from cheapest and most decisive to most expensive. The digest comparison comes first because a skip avoids everything else; the size test comes second because it only matters if there is work to do.
4. Default to the thorough path on failure. The except branch selects TILED, not SKIP. An unreadable header is a reason for caution, and the cost of being wrong in the safe direction is time rather than silence.
5. Record before acting. record_decision runs before the branch executes, so a run that crashes mid-warp still leaves the reason it chose that warp. Recording afterwards loses exactly the cases you most want to explain.
6. Keep the branch bodies interchangeable. warp_single_pass and merge_tiles(warp_tiled(...)) both return a path to a GeoTIFF in the target CRS with the same nodata. Downstream code cannot tell which ran, and should not need to.
7. Return None for a skip rather than raising. A skip is a successful outcome. Raising turns a normal, expected decision into a failure that pollutes the run’s error rate and eventually trains people to ignore it.

Edge Cases & Failure Recovery#

A predicate that is expensive on one source and cheap on the rest. Header reads are fast on a COG and slow on a striped, non-tiled GeoTIFF served over HTTP, where the header may be scattered across the file. If one source’s inspection takes thirty seconds, either cache the inspection result or, better, fix the source’s layout — the same property makes every subsequent read slow too.

Thresholds that drift out of date. TILED_THRESHOLD_MP = 400 encodes an assumption about worker memory. When the worker size changes, the threshold should change with it, and if nobody remembers, the pipeline either tiles unnecessarily or runs out of memory. Derive it from the configured worker memory rather than hard-coding it, and log the derived value at flow start.

A branch that never runs. If ninety-nine per cent of scenes take one path, the other path rots: it is not exercised, its dependencies drift, and the day it is finally selected it fails. Either exercise it in CI with a synthetic input, or delete it and accept the cost of the common path everywhere.

Skips that hide a broken upstream. “Source digest unchanged” is the correct reason to skip, and it is indistinguishable from “the upstream publisher has been down for three weeks”. Track the age of the newest source alongside the skip count, so a run that skips everything because nothing has arrived looks different from a run that skips everything because nothing changed.

A predicate that reads a mutable source twice. Between the inspection and the branch body the source may be republished, so the decision describes a file that no longer exists. Where the source is genuinely mutable, capture its digest at inspection time and pass it forward, then have the branch body verify that the digest it is about to read still matches. That converts a silent mismatch into a cheap, explicit failure that the next run resolves.

Decisions recorded but never read. A decisions table nobody queries is overhead. Put one panel on the pipeline’s dashboard showing the daily distribution of paths; the moment that distribution shifts, someone will want the table, and it will be there.

Branching on a value computed twice. If the predicate reads the size and the branch body reads it again, the two can disagree — the source may have been republished in between. Pass the inspected values forward from the decision rather than re-reading them, which also makes the branch body testable without any I/O at all.

Configuration Reference#

The threshold deserves a little more than a number. It should be derived — worker_memory_gb * 25 megapixels is a reasonable starting formula for a bilinear warp with a 512 MB GDAL cache — and it should be logged at the start of every run so that a change in worker sizing shows up in the logs the first night it takes effect rather than the first night it causes an OOM. Where a pipeline runs on heterogeneous workers, the threshold has to come from the worker actually executing the task, which is an argument for computing it inside the task rather than passing it in from the flow.

Frequently Asked Questions#

Related#

• Branching workflows based on spatial extent — the size predicate in detail
• Skipping tiles with no new source data — the skip branch, and how to record it
• Dynamic task mapping for tile fan-out — the other way a spatial graph changes shape at runtime
• Caching strategies for spatial tasks — where the digest that drives the skip comes from
• Spatial validation & sync tasks — predicates that are validations rather than branches

← Spatial Task Design & Dependency Mapping