Migrating from Prefect to Dagster for Spatial Pipelines

Moving a spatial pipeline from Prefect to Dagster means re-expressing @flow/@task as @job/@op or @asset, translating .map() fan-out into dynamic outputs, porting retry policies onto RetryPolicy, and handing raster and GeoDataFrame IO to Dagster IO managers — then cutting over in phases so both systems run side by side until parity is proven. Migrating from Prefect to Dagster for spatial pipelines is mostly a mechanical mapping of concepts, with the real care reserved for how large geospatial payloads cross task boundaries. This guide gives the concept mapping, a working before/after translation, the IO-manager wiring for spatial types, and a phased cut-over plan. It assumes you have already made the platform decision in Prefect vs Dagster for GIS Workloads and now need to execute the move.

When to Use This Pattern#

A Prefect-to-Dagster migration is worth the effort when:

• You need strict data lineage and versioned materializations for governed spatial data products, not just run history.
• Your mosaics and vector layers are naturally partitioned (by tile, UTM zone, or date) and you want native backfills over those partitions.
• Downstream consumers depend on asset freshness policies and cross-team ownership that Prefect’s task-centric model does not express.
• You are consolidating on Dagster’s asset catalog as the system of record for spatial outputs.

Stay on Prefect if your topology is highly dynamic at runtime — swarms of ad-hoc shapefile uploads or sensor swaths whose count is unknown until execution — where imperative control flow is the natural fit. Migrate for governance, not for novelty.

The single largest conceptual shift is from thinking in runs to thinking in assets. In Prefect you author a flow and reason about what happens when it executes; the outputs are a side effect of the run. In Dagster the reprojected raster, the NDVI layer, and the PostGIS table are the primary objects, and a run is merely the act of bringing one up to date. This inversion pays off precisely for spatial products, because a tiled mosaic partitioned by Web Mercator grid or UTM zone becomes a partitioned asset whose individual cells can be materialized, backfilled, and audited independently — the governance capability that usually motivates the move in the first place.

Concept Mapping#

The table below maps each Prefect concept to its Dagster equivalent, with the spatial consideration that most often bites during translation.

Complete Working Example#

Start from a representative Prefect flow that fans reprojection out over tiles with retries:

# BEFORE — Prefect
from prefect import flow, task
import rasterio
from rasterio.warp import reproject, Resampling, calculate_default_transform
from pyproj import CRS


@task(retries=3, retry_delay_seconds=30, persist_result=True)
def reproject_tile(tile_uri: str, target_crs: str = "EPSG:3857") -> str:
    dst_crs = CRS.from_user_input(target_crs)
    out_uri = tile_uri.replace("/raw/", "/reprojected/")
    with rasterio.open(tile_uri) as src:
        transform, width, height = calculate_default_transform(
            src.crs, dst_crs, src.width, src.height, *src.bounds
        )
        profile = src.profile.copy()
        profile.update(crs=dst_crs, transform=transform,
                       width=width, height=height, nodata=src.nodata)
        with rasterio.open(out_uri, "w", **profile) as dst:
            for b in range(1, src.count + 1):
                reproject(
                    source=rasterio.band(src, b), destination=rasterio.band(dst, b),
                    src_crs=src.crs, dst_crs=dst_crs, dst_transform=transform,
                    resampling=Resampling.bilinear,
                    src_nodata=src.nodata, dst_nodata=src.nodata,
                )
    return out_uri


@flow
def reproject_flow(tile_uris: list[str]) -> list[str]:
    return [f for f in reproject_tile.map(tile_uris)]

The Dagster translation splits the compute from the IO. reproject_array stays a plain function; the @op returns an in-memory array plus profile, and a RasterIOManager handles writing the GeoTIFF. Fan-out becomes a dynamic output:

# AFTER — Dagster
from dagster import (
    op, job, DynamicOut, DynamicOutput, RetryPolicy,
    IOManager, io_manager, OutputContext, InputContext,
)
import numpy as np
import rasterio
from rasterio.warp import reproject, Resampling, calculate_default_transform
from pyproj import CRS


class RasterIOManager(IOManager):
    """Persist (array, profile) tuples as GeoTIFF instead of pickling."""

    def __init__(self, base_uri: str) -> None:
        self.base_uri = base_uri

    def _path(self, context) -> str:
        return f"{self.base_uri}/{'/'.join(context.asset_key.path if context.has_asset_key else context.get_identifier())}.tif"

    def handle_output(self, context: OutputContext, obj: tuple[np.ndarray, dict]) -> None:
        array, profile = obj
        with rasterio.open(self._path(context), "w", **profile) as dst:
            dst.write(array)

    def load_input(self, context: InputContext) -> tuple[np.ndarray, dict]:
        with rasterio.open(self._path(context)) as src:
            return src.read(), src.profile.copy()


@io_manager(config_schema={"base_uri": str})
def raster_io_manager(init_context) -> RasterIOManager:
    return RasterIOManager(init_context.resource_config["base_uri"])


@op(out=DynamicOut(str))
def enumerate_tiles(tile_uris: list[str]):
    for uri in tile_uris:
        key = uri.rsplit("/", 1)[-1].replace(".tif", "")
        yield DynamicOutput(uri, mapping_key=key)


@op(retry_policy=RetryPolicy(max_retries=3, delay=30),
    io_manager_key="raster_io")
def reproject_tile_op(tile_uri: str, target_crs: str = "EPSG:3857") -> tuple[np.ndarray, dict]:
    dst_crs = CRS.from_user_input(target_crs)
    with rasterio.open(tile_uri) as src:
        transform, width, height = calculate_default_transform(
            src.crs, dst_crs, src.width, src.height, *src.bounds
        )
        profile = src.profile.copy()
        profile.update(crs=dst_crs, transform=transform,
                       width=width, height=height, nodata=src.nodata)
        out = np.empty((src.count, height, width), dtype=src.dtypes[0])
        for b in range(src.count):
            reproject(
                source=rasterio.band(src, b + 1), destination=out[b],
                src_crs=src.crs, dst_crs=dst_crs, dst_transform=transform,
                resampling=Resampling.bilinear,
                src_nodata=src.nodata, dst_nodata=src.nodata,
            )
    return out, profile


@job(resource_defs={"raster_io": raster_io_manager})
def reproject_job():
    enumerate_tiles().map(reproject_tile_op)

For vector stages, write a sibling GeoDataFrameIOManager whose handle_output calls gdf.to_parquet(path) and whose load_input calls gpd.read_parquet(path), preserving CRS through the GeoParquet metadata rather than serializing a GeoDataFrame with pickle. Centralizing raster and vector persistence in IO managers is what keeps the ported ops clean and aligned with the DAG Design Principles for Spatial ETL — the compute never touches storage paths directly. The dynamic-output fan-out maps one-to-one onto the tiling approach in parametrizing spatial DAGs by tile index.

Phased Cut-Over Plan#

Do not flip a production spatial pipeline in one release. Move in stages:

1. Inventory and freeze. Catalog every Prefect flow, task, retry policy, block, and result store. Freeze new feature work on the Prefect side so you are porting a stable target.
2. Port leaf compute first. Translate pure per-tile tasks (reprojection, band math) into Dagster ops with IO managers. These have the fewest dependencies and are easy to verify against Prefect outputs byte-for-byte.
3. Rebuild the graph as assets. Express the lineage — raw tile → reprojected → index → mosaic → PostGIS — as software-defined assets with partitions matching your tiling scheme.
4. Run both in shadow. Execute Dagster in parallel with Prefect on the same inputs, comparing outputs with gdalinfo checksums and feature counts. Route no consumers to Dagster yet.
5. Cut over reads, then writes. Point downstream consumers at the Dagster-materialized assets, keep Prefect as a warm fallback for one cycle, then decommission the Prefect deployment.

The shadow-run phase is where spatial migrations either build confidence or expose silent drift, so make the parity check pixel-exact rather than impressionistic. For raster assets, compare the two engines’ outputs with a checksum over the pixel array plus an equality assertion on crs, transform, and nodata; a mosaic that looks right in a viewer but has a one-pixel offset in its geotransform will corrupt every downstream spatial join. For vector assets, compare feature counts, the bounding box of the union, and the set of EPSG codes present. Automate this as a nightly job that diffs Dagster materializations against the still-running Prefect outputs and fails loudly on any mismatch, so you cut over on evidence instead of optimism.

Retry and state semantics differ subtly between the two engines; review Prefect flow state transitions explained before assuming a RetryPolicy reproduces your old backoff, and lean on checkpointing large raster mosaics so long builds survive the shadow-run phase on either engine. Consult the official Dagster IO managers documentation for the current resource-config API.

Common Pitfalls#

• Pickling GeoDataFrames and rasters between steps. Dagster’s default IO manager serializes return values with pickle, which is slow and CRS-lossy for spatial types. Always register raster and vector IO managers that round-trip through GeoTIFF and GeoParquet.
• UUID mapping keys. Porting .map() to DynamicOut with a random mapping_key breaks retries and backfills — the key must be a stable function of the tile index, exactly as in the fan-out companion page.
• Assuming identical retry timing. Prefect’s retry_delay_seconds list and Dagster’s RetryPolicy(delay=...) are not the same backoff shape. Validate against your rate-limit budget rather than trusting a literal number-for-number copy.
• Losing CRS at IO boundaries. Reading a GeoParquet or GeoTIFF back without confirming crs and nodata reintroduces the projection ambiguity the pipeline was built to prevent. Assert both on load inside the IO manager.

Related: Decide direction with Prefect vs Dagster for GIS Workloads, translate fan-out via parametrizing spatial DAGs by tile index, keep long migrations resumable with checkpointing large raster mosaics, and understand engine state via Prefect flow state transitions explained.

← Back to Prefect vs Dagster for GIS Workloads