Migrating From Prefect to Dagster for Spatial Pipelines

The migration that works is not a rewrite; it is an extraction followed by a series of small cutovers. Pull the spatial work out of the task decorators into plain functions, wrap those in Dagster assets, seed the partition record from your existing ledger, and move one layer at a time with both control planes running. Every step leaves a working pipeline, and any of them can be the last one if you change your mind.

When to Use This Pattern#

• The ledger has become the largest part of the codebase, which is the usual signal that the model is fighting the work.
• Backfills happen weekly and each one needs a hand-written manifest filter.
• Non-engineers operate the pipeline and need a partition view rather than a run list.
• The grid is stable — a service area, a national pyramid — rather than decided per run.

Complete Working Example#

Step one is the one that does the real work, and it involves neither tool. The spatial logic moves out of the task body into a function that knows nothing about orchestration:

# domain/tiles.py — no orchestrator imports anywhere in this module.
from __future__ import annotations

from dataclasses import dataclass


@dataclass(frozen=True)
class TileResult:
    path: str
    valid_fraction: float
    seconds: float
    source_digest: str


def build_tile(tile: Tile, source: SourceSpec, dest_root: str) -> TileResult:
    """Everything that was inside the @task body, and nothing else."""
    window = fetch_window(source, tile)
    stats = warp_and_write(window, tile, dest_root)
    return TileResult(stats.path, stats.valid_fraction, stats.seconds, source.digest)

Both orchestrators then become thin wrappers. The Prefect one shrinks to a decorator plus the ledger it already maintained:

@task(retries=2, timeout_seconds=900, tags=["tile"])
def build_tile_task(tile: Tile, source: SourceSpec, dest: str) -> TileResult:
    key = work_key(tile, source.digest, RECIPE)
    if ledger_has(key):
        return ledger_get(key)
    result = build_tile(tile, source, dest)          # the extracted function
    ledger_mark(key, tile, result)
    return result

and the Dagster one calls the same function, with the materialisation replacing the ledger write:

import dagster as dg

TILES = dg.StaticPartitionsDefinition(
    [f"{z}_{x}_{y}" for z, x, y in tiles_intersecting(SERVICE_AREA_WKT, 12)]
)


@dg.asset(partitions_def=TILES, pool="scene-reads",
          retry_policy=dg.RetryPolicy(max_retries=2, delay=15))
def warped_tile(context: dg.AssetExecutionContext) -> dg.MaterializeResult:
    z, x, y = (int(v) for v in context.partition_key.split("_"))
    result = build_tile(Tile(z, x, y), SOURCE, DEST)   # the same function
    return dg.MaterializeResult(metadata={
        "path": dg.MetadataValue.path(result.path),
        "valid_px_pct": dg.MetadataValue.float(result.valid_fraction * 100),
        "source_digest": dg.MetadataValue.text(result.source_digest),
    })

Seeding the partition record from the existing ledger is what avoids rebuilding everything on the first Dagster run:

def seed_materialisations(instance: dg.DagsterInstance, rows) -> int:
    """One AssetMaterialization per ledger row, so day one is not a full backfill."""
    for row in rows:                                   # SELECT z, x, y, ... FROM tile_ledger
        instance.report_runless_asset_event(
            dg.AssetMaterialization(
                asset_key=dg.AssetKey("warped_tile"),
                partition=f"{row.z}_{row.x}_{row.y}",
                metadata={"seeded_from": "tile_ledger",
                          "source_digest": dg.MetadataValue.text(row.source_digest)},
            )
        )
    return len(rows)

Seeding deserves emphasis because skipping it is the commonest way a migration becomes expensive. Without seeded materialisations, Dagster’s view of the world on day one is that nothing exists, so the first scheduled run treats a fully-built national pyramid as a complete backfill. That is days of compute and a very large bill for information you already had in a table. The seeding script is thirty lines and turns the cutover into a no-op run.

Parameter & Option Reference#

Verification & Testing#

The tests that matter compare the two implementations rather than testing either alone.

def test_domain_function_has_no_orchestrator_imports() -> None:
    src = Path("domain/tiles.py").read_text()
    assert "prefect" not in src and "dagster" not in src


def test_both_wrappers_produce_identical_output(tmp_path) -> None:
    a = run_prefect_tile(TILE, dest=tmp_path / "p")
    b = run_dagster_tile(TILE, dest=tmp_path / "d")
    assert sha256_of(a) == sha256_of(b), "the wrappers are not calling the same code"


def test_seeding_maps_every_ledger_row(db, instance) -> None:
    rows = db.all("SELECT z, x, y, source_digest FROM tile_ledger WHERE layer='ortho'")
    seed_materialisations(instance, rows)
    seen = instance.get_materialized_partitions(dg.AssetKey("warped_tile"))
    assert seen == {f"{r.z}_{r.x}_{r.y}" for r in rows}


def test_partition_keys_match_the_ledger_shape(db) -> None:
    # A single mismatched key format is a full backfill nobody asked for.
    sample = db.one("SELECT z, x, y FROM tile_ledger LIMIT 1")
    assert f"{sample.z}_{sample.x}_{sample.y}" in TILES.get_partition_keys()

The last test is small and prevents the most expensive failure in the list. A ledger storing 12/2147/1398 and a partition set keyed 12_2147_1398 will seed nothing at all, silently, because every reported partition falls outside the definition — and the first run then rebuilds the world. Asserting one real row against the real partition set costs a millisecond and catches the whole class.

The dual-write window is worth planning rather than improvising. For one full cycle after a layer cuts over, keep writing the old ledger from the Dagster wrapper as well — a few extra lines in the asset body — so that rolling that layer back means re-enabling a schedule and nothing else. Without it, a rollback after three days means the old ledger is three days stale, and the Prefect deployment’s first run rebuilds everything it thinks is missing. That is the same phantom backfill the seeding step exists to prevent, arriving from the other direction and at a worse moment.

The other thing worth doing during the window is comparing the two records rather than trusting either. A nightly query that diffs the ledger’s built set against Dagster’s materialised partitions will surface a key-format mismatch, a missed seeding batch or a layer that is quietly being built by both planes — all of which are cheap to fix while both systems are running and awkward afterwards. The diff should be empty every night; the day it is not is the day the migration told you something.

Common Pitfalls#

• Rewriting the spatial code during the migration. Two variables change at once and any output difference becomes unattributable. Extract first, migrate second.
• Skipping the seeding step. The first run becomes a full backfill of work already done.
• Mismatched partition key format. Seeding silently does nothing and the same full backfill follows.
• Cutting over every layer at once. A rollback then means the whole product, at the worst possible moment.
• Decommissioning early. Keep the old deployment until the new one has survived a backfill and a real recovery, not merely a week of green runs.
• Sharing one database between the control planes. It saves nothing and makes a clean rollback impossible.

Frequently Asked Questions#

Related#

• Prefect vs Dagster for GIS workloads — deciding whether to migrate at all
• Mapping tasks over a tile grid in Dagster — the destination shape
• State management in geospatial flows — the ledger being seeded from
• Prefect flow state transitions explained — what you are leaving behind
• Testing spatial flows in CI with synthetic fixtures — how the extracted functions get tested

← Prefect vs Dagster for GIS Workloads