Scheduling Spatial Validation with Dagster Assets

If you think about your pipeline in terms of the datasets it produces — “the validated parcels layer for each tile” rather than “the 2 a.m. job” — Dagster’s software-defined assets are a natural fit. This page shows how to model a validated spatial layer as a partitioned asset, attach asset checks that assert geometry validity, and trigger recomputation with schedules and sensors, all with runnable code. It assumes single-deployment Dagster up to a few thousand tile partitions and sits under orchestration tool selection for spatial pipelines, which frames the broader choice between Dagster, Airflow, and Prefect. The actual geometry logic here reuses the checks from the rule engines built with GeoPandas pattern, so Dagster stays a scheduling and observability layer, not a home for spatial code.

Prerequisites

• Python 3.10+ with Dagster 1.7.x: pip install "dagster==1.7.*" "dagster-webserver==1.7.*". Launch the local UI with dagster dev.
• GeoPandas 0.14+ / Shapely 2.0+ / PyArrow 14+ for reading and writing partitioned GeoParquet. Gotcha: install dagster and its extras in a fresh environment; Dagster pins pydantic v2 and will conflict with an Airflow environment.
• A source dataset with a stable partition key. Tiles must map to a fixed set of keys (for example a grid identifier); a StaticPartitionsDefinition or DynamicPartitionsDefinition needs those keys to exist before materialisation.
• Object storage or a local base path for the validated output; asset input/output managers read and write there, and geometry never travels through Dagster’s event log.

Step-by-Step Procedure

Step 1 — Define the validated layer as a partitioned asset

Model the validated output as one asset partitioned by tile. The asset function reads the source, normalises the coordinate reference system (CRS), and writes validated GeoParquet.

# defs.py
import geopandas as gpd
from dagster import asset, StaticPartitionsDefinition, MaterializeResult, MetadataValue

TILES = StaticPartitionsDefinition(["tile_0001", "tile_0002", "tile_0003"])
BASE = "/data/validated"


@asset(partitions_def=TILES, group_name="spatial_validation")
def validated_parcels(context) -> MaterializeResult:
    tile = context.partition_key
    gdf = gpd.read_file(f"/data/source/{tile}.gpkg", layer="parcels")
    if gdf.crs is None:
        raise ValueError(f"{tile}: source layer has no CRS")
    gdf = gdf.to_crs(epsg=4326)

    invalid = int((~gdf.geometry.is_valid).sum())
    out = f"{BASE}/{tile}.parquet"
    gdf.to_parquet(out)                      # overwrite: idempotent per partition

    return MaterializeResult(metadata={
        "tile": tile,
        "features": len(gdf),
        "invalid": MetadataValue.int(invalid),
        "path": out,
    })

Verification: run dagster dev, then materialise a single partition from the UI; confirm the asset’s metadata shows the feature count and an invalid value, and the GeoParquet file appears on disk.

Step 2 — Attach asset checks for geometry validity

Asset checks bind your validation rules to the asset so a failure blocks downstream materialisation and is recorded in the asset graph.

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


@asset_check(asset=validated_parcels, blocking=True)
def geometry_is_valid(context) -> AssetCheckResult:
    tile = context.partition_key
    gdf = gpd.read_parquet(f"{BASE}/{tile}.parquet")
    invalid = int((~gdf.geometry.is_valid).sum())
    return AssetCheckResult(
        passed=(invalid == 0),
        severity=AssetCheckSeverity.ERROR,
        metadata={"invalid_geometries": invalid},
    )


@asset_check(asset=validated_parcels)
def crs_is_canonical(context) -> AssetCheckResult:
    tile = context.partition_key
    gdf = gpd.read_parquet(f"{BASE}/{tile}.parquet")
    return AssetCheckResult(
        passed=(gdf.crs is not None and gdf.crs.to_epsg() == 4326),
        severity=AssetCheckSeverity.WARN,
        metadata={"epsg": None if gdf.crs is None else gdf.crs.to_epsg()},
    )

Verification: materialise a partition containing a known self-intersecting polygon; confirm geometry_is_valid reports passed=False with severity ERROR and blocks dependants, while a clean tile passes both checks.

Step 3 — Schedule the asset and add a freshness sensor

A schedule keeps partitions fresh on a cadence; a sensor materialises a tile the moment its source file arrives.

from dagster import (
    define_asset_job, ScheduleDefinition, sensor, RunRequest,
    SensorEvaluationContext,
)
import os

validate_job = define_asset_job("validate_job", selection=[validated_parcels])

daily_schedule = ScheduleDefinition(
    job=validate_job,
    cron_schedule="0 2 * * *",
)


@sensor(job=validate_job)
def new_tile_sensor(context: SensorEvaluationContext):
    """Request a run for each source file that has appeared since the last tick."""
    for fname in os.listdir("/data/source"):
        if fname.endswith(".gpkg"):
            tile = fname.removesuffix(".gpkg")
            yield RunRequest(run_key=fname, partition_key=tile)

Verification: drop a new tile_0004.gpkg into /data/source, wait one sensor tick, and confirm Dagster launches a run for the tile_0004 partition without manual action.

Step 4 — Register definitions and expose them

from dagster import Definitions

defs = Definitions(
    assets=[validated_parcels],
    asset_checks=[geometry_is_valid, crs_is_canonical],
    jobs=[validate_job],
    schedules=[daily_schedule],
    sensors=[new_tile_sensor],
)

Verification: run dagster dev and confirm the asset graph shows validated_parcels with two attached checks, the schedule appears under Automation, and the sensor is listed and can be toggled on.

Interpreting Results

Dagster records every materialisation and every asset-check result as a timestamped event, which turns data quality into a browsable history rather than a transient log line. Read results at three levels:

Because a blocking check stops downstream assets, a failed geometry_is_valid on one tile prevents a stale or invalid layer from propagating into a published map, while unaffected tiles continue — the asset-graph equivalent of per-tile failure isolation.

Backfills over partitions. When a validation rule tightens — a stricter precision tolerance, a new topology check — you must re-validate history. Dagster makes this a single action: select the validated_parcels asset, choose a range of tile partitions, and launch a backfill from the UI or with dagster asset backfill --partitions tile_0001...tile_0100. Each partition re-materialises independently and re-runs its asset checks, so the backfill both refreshes the data and produces a fresh quality record for every reprocessed tile. Because each partition is idempotent — the asset overwrites its GeoParquet keyed on the tile — a backfill that overlaps a scheduled run cannot duplicate features. This is the same reprocessing need that Airflow serves with dags backfill, expressed here in terms of the dataset rather than a date-stamped job.

Freshness signals downstream work. A materialised, check-passing partition is itself a data-aware signal: any asset that declares validated_parcels as an upstream dependency is scheduled to recompute when its tiles refresh. This converts the fragile “publish runs at 03:00 and hopes validation finished” timing assumption into an explicit lineage edge in the asset graph, so a published-tiles asset rebuilds exactly the tiles whose validated inputs changed and nothing more.

Gotchas & Edge Cases

Asset checks re-read the materialised output. The checks in Step 2 read the GeoParquet the asset just wrote. Do not re-run the full validation inside the check; read the persisted result and assert on it, or you double the compute per tile.

Partition keys must exist before the sensor references them. With DynamicPartitionsDefinition, register a new key with context.instance.add_dynamic_partitions(...) before issuing a RunRequest for it, or the run request is dropped silently.

Blocking checks need blocking=True and a selection that includes them. A non-blocking check records a failure but still lets dependants materialise. Set blocking=True on any check whose failure must stop publication.

Geometry does not belong in the event log. Return small metadata values (counts, paths, EPSG codes) from MaterializeResult; never attach a serialised GeoDataFrame as metadata, which would bloat the Dagster instance database.

Sensor cursors prevent duplicate runs — use run_key. The run_key=fname in Step 3 deduplicates: the same file will not launch two runs. Omitting it re-triggers the tile on every tick.

Input/output managers can hide the reference-not-geometry rule. Dagster’s default I/O manager will pickle whatever an asset returns. If you return a GeoDataFrame from the asset function, Dagster serialises the entire geometry to the I/O manager’s store on every materialisation — the same anti-pattern as pushing geometry through XCom in Airflow. Return a lightweight MaterializeResult with metadata as shown in Step 1, write the GeoParquet yourself to a known path, and let downstream assets read that path. Reserve typed I/O managers for the small summary objects, not for the layers themselves.

Schedules and sensors are off by default. After registering Definitions, both the daily_schedule and new_tile_sensor start in a stopped state. They must be toggled on in the UI or set to DefaultScheduleStatus.RUNNING / DefaultSensorStatus.RUNNING in code, or the pipeline will appear correctly wired yet never fire automatically.

When to Escalate

• You need heavy fan-out across a compute cluster: Dagster schedules the work but is not itself a distributed geometry engine; delegate per-partition compute to Dask or Spark as covered in batch processing large spatial datasets.
• A team standardised on Airflow or Prefect already: switching an established platform is costly; weigh the trade-offs in Airflow vs Prefect for spatial pipeline orchestration before migrating.
• Inputs are truly streaming: sensors poll on a tick interval, so sub-second latency needs a queue instead — see asynchronous validation workflows.

Frequently Asked Questions

Related

• Orchestration Tool Selection for Spatial Pipelines — where Dagster fits among scheduling models and when to prefer it
• Airflow vs Prefect for Spatial Pipeline Orchestration — the imperative-scheduler alternatives to Dagster’s asset model
• Building Rule Engines with GeoPandas — the geometry validation logic that asset checks invoke

Back to Orchestration Tool Selection for Spatial Pipelines