This guide is part of Orchestrating Spatial ETL Pipelines, the reference for turning extract, clean, and load scripts into auditable, restartable spatial data products.

Managing Spatial Assets with Dagster

Most orchestrators ask you to describe tasks: run this function, then that one, in this order. Dagster asks you to describe assets — the durable objects your pipeline produces, such as a reprojected scene, a validated tile, or a loaded PostGIS table — and infers the execution graph from how those objects depend on one another. For spatial ETL this reframing is unusually productive, because geospatial data is already indexed by natural keys — a scene date, a tile ID, a UTM zone — that map cleanly onto Dagster’s partitioning model. A satellite archive is not a stream of anonymous task runs; it is a materialized grid of tiles and dates, and a software-defined asset graph lets you address, test, and rebuild any cell of that grid in isolation.

The naive alternative — a chain of @ops or a plain Python script that reprojects, validates, and loads in one pass — fails the same way every scaled spatial pipeline fails. A single corrupt scene aborts the whole run; a re-run reprocesses six months of imagery to fix one bad tile; and there is no record of which outputs are current versus stale. Modeling each stage as an asset with an explicit partition scheme converts those failures into targeted, cheap operations: a bad tile is one partition to rematerialize, a quality regression is one asset check that fails loudly, and lineage is a first-class graph the UI renders for you.

Prerequisites & Environment

The asset APIs below target the modern Dagster surface (dagster.asset, AssetCheckResult, typed ConfigurableResource/ConfigurableIOManager). Pin the framework and the geospatial stack together — Dagster’s partition and check APIs moved quickly across minor releases, and GeoPandas 1.0 changed the default GeoParquet writer.

• Python 3.10+ — required for modern type hints and current dagster>=1.7 wheels.
• Dagster — dagster>=1.7,<1.9 for general-availability asset checks; see the matrix below before pinning older.
• Geospatial stack — geopandas>=1.0, shapely>=2.0 (vectorized geometry predicates), pyproj>=3.6, pyarrow>=14 for GeoParquet.
• Load target — SQLAlchemy>=2.0 and psycopg2-binary for the PostGIS resource; boto3 for the S3-backed IO manager.

pip install "dagster>=1.7,<1.9" dagster-webserver \
  "geopandas>=1.0" "shapely>=2.0" "pyproj>=3.6" "pyarrow>=14" \
  "SQLAlchemy>=2.0" psycopg2-binary boto3

Verify the install resolves the asset-check symbols you depend on before writing any asset:

python -c "import dagster; from dagster import asset, asset_check, AssetCheckResult; print(dagster.__version__)"

If that import raises ImportError on asset_check, you are on a release older than 1.5 and the quality-gate pattern in this guide will not apply.

Version / Compatibility Matrix

Pin dagster and dagster-webserver to the same version; the UI reads check and partition metadata the daemon writes, and a mismatch renders partial or empty asset-check panels.

Step-by-Step Implementation

Step 1 — Define the raw-to-loaded asset graph

The core idea: each durable spatial representation is one @asset, and Dagster reads the dependency edges from the function signatures. reprojected_scenes takes raw_scenes as an argument, so Dagster knows it depends on it — no explicit DAG wiring.

from dagster import asset, AssetExecutionContext
import geopandas as gpd

@asset(io_manager_key="geoparquet_io", group_name="scenes")
def raw_scenes(context: AssetExecutionContext) -> gpd.GeoDataFrame:
    """Ingest the raw footprints for one materialization as a GeoDataFrame."""
    gdf = gpd.read_file(context.op_config["source_uri"])  # placeholder ingest
    context.add_output_metadata({"rows": len(gdf), "crs": str(gdf.crs)})
    return gdf


@asset(io_manager_key="geoparquet_io", group_name="scenes")
def reprojected_scenes(raw_scenes: gpd.GeoDataFrame) -> gpd.GeoDataFrame:
    """Warp incoming footprints onto the pipeline's canonical analysis CRS."""
    target_crs: str = "EPSG:3857"
    return raw_scenes.to_crs(target_crs)

The dependency between raw_scenes and reprojected_scenes is implicit and typed. Raster reprojection follows the same principle as vector .to_crs() but is governed by resampling and grid-alignment rules covered in raster alignment and resampling techniques; wrap that warp step inside reprojected_scenes when the asset carries imagery rather than footprints.

Step 2 — Persist geometry with a GeoParquet IOManager

Returning a GeoDataFrame from an asset is convenient, but you must not let Dagster pickle it into its default filesystem store — pickled geometry is fragile across library versions and useless to other tools. A custom ConfigurableIOManager serializes each asset to GeoParquet in object storage and reads it back on demand, keyed by asset name and partition.

from dagster import ConfigurableIOManager, InputContext, OutputContext
import geopandas as gpd

class GeoParquetIOManager(ConfigurableIOManager):
    """Persist each spatial asset as GeoParquet under s3://<bucket>/<asset>/<partition>."""
    base_uri: str

    def _path(self, context) -> str:
        partition = context.asset_partition_key if context.has_partition_key else "all"
        asset_name = "/".join(context.asset_key.path)
        return f"{self.base_uri}/{asset_name}/{partition}.parquet"

    def handle_output(self, context: OutputContext, gdf: gpd.GeoDataFrame) -> None:
        uri = self._path(context)
        gdf.to_parquet(uri, index=False)  # GeoParquet via pyarrow
        context.add_output_metadata({"uri": uri, "rows": len(gdf)})

    def load_input(self, context: InputContext) -> gpd.GeoDataFrame:
        return gpd.read_parquet(self._path(context.upstream_output))

Because the payload lives in object storage, the same asset graph runs identically under the in-process executor during tests and the multiprocess or Kubernetes launcher in production — the geometry never has to survive a worker boundary in memory.

Step 3 — Partition assets by date and tile

Partitioning is where spatial data and Dagster fit together. A DailyPartitionsDefinition names one partition per acquisition date; a StaticPartitionsDefinition names one per tile in a fixed grid; a MultiPartitionsDefinition names the Cartesian product of the two.

from dagster import (
    asset,
    AssetExecutionContext,
    DailyPartitionsDefinition,
    StaticPartitionsDefinition,
    MultiPartitionsDefinition,
)
import geopandas as gpd

scene_grid = MultiPartitionsDefinition(
    {
        "date": DailyPartitionsDefinition(start_date="2024-01-01"),
        "tile": StaticPartitionsDefinition(["T31UDQ", "T31UEQ", "T32ULV"]),
    }
)

@asset(partitions_def=scene_grid, io_manager_key="geoparquet_io", group_name="scenes")
def reprojected_scenes(context: AssetExecutionContext) -> gpd.GeoDataFrame:
    """Materialize exactly one (date, tile) cell from the partition key."""
    keys = context.partition_key.keys_by_dimension
    date, tile = keys["date"], keys["tile"]
    context.log.info("Processing tile=%s date=%s", tile, date)
    gdf = gpd.read_parquet(f"s3://raw/{tile}/{date}.parquet")
    return gdf.to_crs("EPSG:3857")

Materializing a single cell reads and writes only that tile-date, which is the whole point: the tile-partitioned asset pattern lets each tile track its own freshness, and the scene-partition backfill workflow reprocesses only the date range that changed.

Step 4 — Add asset checks as spatial quality gates

An asset check is a named assertion about an asset’s data that Dagster surfaces in the UI and — on 1.7+ with blocking=True — uses to stop a bad output from propagating. For spatial data the two highest-value checks are CRS conformance and geometry validity, both expressible with the Shapely 2.0 vectorized predicates.

from dagster import asset_check, AssetCheckResult, AssetCheckSeverity
import geopandas as gpd
import shapely

@asset_check(asset="reprojected_scenes", blocking=True)
def crs_is_canonical(reprojected_scenes: gpd.GeoDataFrame) -> AssetCheckResult:
    """Block downstream load if the asset is not on the analysis CRS."""
    ok = reprojected_scenes.crs is not None and reprojected_scenes.crs.to_epsg() == 3857
    return AssetCheckResult(
        passed=bool(ok),
        severity=AssetCheckSeverity.ERROR,
        metadata={"epsg": reprojected_scenes.crs.to_epsg() if reprojected_scenes.crs else None},
    )


@asset_check(asset="reprojected_scenes")
def geometry_is_valid(reprojected_scenes: gpd.GeoDataFrame) -> AssetCheckResult:
    """Warn on any invalid geometry using the vectorized Shapely 2.0 predicate."""
    valid_mask = shapely.is_valid(reprojected_scenes.geometry.values)
    invalid = int((~valid_mask).sum())
    return AssetCheckResult(
        passed=invalid == 0,
        severity=AssetCheckSeverity.WARN,
        metadata={"invalid_geometries": invalid, "total": len(reprojected_scenes)},
    )

On Dagster 1.5 and 1.6 the blocking=True flag is not fully honored, so replicate the CRS assertion inside loaded_scenes as a defensive guard. On 1.7+ the blocking check alone is sufficient to stop loaded_scenes from materializing when the CRS is wrong.

Step 5 — Wire resources and trigger with sensors and schedules

Resources inject external systems — a PostGIS engine, an S3 client — into assets without hard-coding connection strings. A schedule fires the daily partition on a cron cadence; a sensor fires when new source data appears.

from dagster import (
    ConfigurableResource,
    Definitions,
    define_asset_job,
    build_schedule_from_partitioned_job,
)
from sqlalchemy import create_engine, Engine

class PostGISResource(ConfigurableResource):
    """A reusable PostGIS connection wired into any loading asset."""
    dsn: str

    def get_engine(self) -> Engine:
        return create_engine(self.dsn, pool_pre_ping=True)


scenes_job = define_asset_job("scenes_job", selection="scenes", partitions_def=scene_grid)
daily_schedule = build_schedule_from_partitioned_job(scenes_job)

defs = Definitions(
    assets=[raw_scenes, reprojected_scenes],
    asset_checks=[crs_is_canonical, geometry_is_valid],
    jobs=[scenes_job],
    schedules=[daily_schedule],
    resources={
        "geoparquet_io": GeoParquetIOManager(base_uri="s3://spatial-lake/assets"),
        "postgis": PostGISResource(dsn="postgresql+psycopg2://etl@db/gis"),
    },
)

build_schedule_from_partitioned_job derives a cron schedule that targets the newest partition automatically — for a daily grid it launches yesterday’s date at each tick, so the schedule and the partition definition never drift out of sync.

Advanced Patterns & Edge Cases

Per-tile independence and freshness

When each tile is its own partition, Dagster tracks materialization state per cell, so the UI shows exactly which tiles are stale after a grid change. This is the foundation for treating a raster mosaic as a live product rather than a periodic rebuild — see partitioning spatial assets by tile in Dagster for the StaticPartitionsDefinition and custom-tile-grid recipe, including reading context.partition_key to process a single tile.

Historical reprocessing without a full rebuild

A CRS-registration fix or an upstream algorithm change means old scenes must be recomputed — but only the affected window. Dagster’s partition backfills recompute only the partitions in the requested range and parallelize them across the run launcher, which is covered end to end in backfilling satellite scene partitions in Dagster, including partition concurrency limits and failed-partition retries.

Incremental windows and watermarks

Schedules that always target “yesterday” are simple but brittle if a run is missed. Pairing a DailyPartitionsDefinition with a watermark of the last successfully materialized partition lets a catch-up run fill gaps deterministically — the discipline is developed in scheduling and incremental spatial loads, which complements Dagster’s built-in partition cursor.

Performance Optimization

The default in-process executor materializes partitions serially. For a backfill across many tiles the win is concurrency, tuned to the slowest external dependency — usually the source API or the PostGIS write path, not the CPU. Set a run-level concurrency cap and let Dagster schedule partition runs against it.

from dagster import define_asset_job

# Cap concurrent partition runs so a 200-tile backfill does not open
# 200 PostGIS connections at once. Benchmarked: 8 workers sustained
# ~46 tiles/min against a single db; 32 workers saturated the write
# lock and dropped to ~28 tiles/min. The knee was at 8-12.
scenes_job = define_asset_job(
    "scenes_job",
    selection="scenes",
    config={"execution": {"config": {"multiprocess": {"max_concurrent": 8}}}},
)

The measured lesson generalizes: for spatial loads the optimum concurrency is set by the target’s write contention, so benchmark against your PostGIS instance rather than assuming more workers is faster. Tag the loading asset with a concurrency pool so it never exceeds the connection budget even when other assets run alongside it.

Integration into ETL Pipelines

The asset graph is the schema contract: each asset’s add_output_metadata call records row counts, CRS, and the GeoParquet URI, giving downstream consumers a typed, queryable lineage instead of opaque file drops. Route quality-gate failures the same way the rest of the pipeline does — a non-blocking geometry_is_valid check that reports 40 invalid geometries should feed a dead-letter partition rather than aborting the run, so the other tiles still load. Wire the PostGISResource into a final loaded_scenes asset that upserts on scene_id, keeping the write idempotent so a rematerialization overwrites rather than appends. Emit the partition key in every log line so an operator can answer “did tile T31UDQ for 2024-06-01 load?” from the Dagster UI without opening a database session.

Troubleshooting Reference

Related

• Partitioning Spatial Assets by Tile in Dagster — StaticPartitions and custom tile grids so each tile materializes and tracks independently
• Backfilling Satellite Scene Partitions in Dagster — targeted, parallel reprocessing of a historical date window with partition retries
• Orchestrating Spatial ETL Pipelines — the reference model for spatial DAGs, idempotency, and load-target selection
• Scheduling and Incremental Spatial Loads — watermark tracking and catch-up runs that complement Dagster partitions
• Raster Alignment and Resampling Techniques — the reprojection and grid-alignment rules that live inside the reprojected_scenes asset