Geospatial Data Ingestion Pipelines

Ingestion is the stage where an energy-GIS pipeline earns or loses its determinism, and it is the first stage of the Core Energy-GIS Data & Spatial Fundamentals architecture. A renewable portfolio pulls parcel boundaries as Shapefiles, transmission corridors as GeoPackages, interconnection-queue exports as GeoParquet, jurisdictional feeds as GeoJSON, utility asset registers from PostGIS, and national resource layers streamed straight from s3://, gs://, or az:// object storage. Every one of those sources has its own column names, its own geometry encoding, and its own idea of whether a coordinate reference system is worth declaring. The specific failure mode this page addresses is contract leakage: a source that does not match the shape your pipeline assumes slips past ingestion unchecked, then silently corrupts a spatial join, a capacity calculation, or a compliance overlay several stages later — far from the line of code that actually let it in.

This page builds an ingestion layer whose entire job is to refuse ambiguity at the door. Heterogeneous inputs are read, validated against an explicit schema, assigned a deterministic key so a replayed batch overwrites rather than duplicates, and emitted as one predictable GeoDataFrame contract that every downstream stage can trust. Ingestion here is deliberately narrow: it does not reproject and it does not repair topology — it records the native coordinate frame and geometry validity as facts, then hands a clean, uniform frame off to the projection and quality stages that specialise in those transforms. That separation is what keeps each stage independently testable and keeps a bad row from masquerading as a good one.

The Ingestion Contract: Validate at the Boundary

The single principle that makes ingestion deterministic is validate at the boundary: the moment external data crosses into your process, it is coerced to a declared schema or rejected, and nothing past that line is allowed to defensively re-check what the boundary already guaranteed. Ad-hoc ingestion fails for four compounding reasons, and only the last of them reliably raises an exception where the mistake is actually made.

First, schema drift. A provider renames cap_mw to capacity_mw, changes voltage_kv from a string to a float, or ships a new nullable column. A script that reads columns positionally or with .get() fallbacks absorbs the change without complaint and produces subtly wrong numbers downstream.

Second, undeclared or heterogeneous CRS. Shapefiles arrive with a .prj, GeoJSON is assumed to be EPSG:4326 but frequently is not, and a PostGIS export can carry any SRID. If ingestion does not record each source’s coordinate frame as an explicit attribute, the downstream coordinate reference system alignment stage has nothing reliable to transform from.

Third, non-idempotent loading. Re-running a job after a partial failure — the normal case when streaming hundreds of gigabytes over an unreliable link — appends the batch a second time. Without a stable, content-derived key, the target grows a duplicate of every already-loaded asset, and every capacity sum after that is inflated.

Fourth, partial writes. A worker killed mid-write leaves a half-serialised GeoParquet file that reads back as a truncated or corrupt frame. This is the one failure that usually does surface as an exception, but only when something later tries to read the wreckage.

The diagram below shows the boundary as a single gate: heterogeneous sources enter, the contract is enforced once, and only records that satisfy it become the canonical frame handed to CRS alignment and topology repair. Everything that fails is quarantined with its reason, never silently dropped.

Because the gate runs once, downstream stages stop defensively re-validating. That is not a stylistic nicety — a distance calculation that re-checks CRS, a raster join that re-checks geometry validity, and a compliance overlay that re-checks column types are three places the same bug can hide. Ingestion’s contract collapses that surface area to one.

Prerequisites & Data Requirements

This workflow assumes the following inputs and environment:

Source assets as vector features — Point substations and generation sites, LineString/MultiLineString conductors, or Polygon parcels — carrying at minimum an asset_id, a capacity_mw, and a defined, non-null CRS. Public sources should be drawn from versioned, machine-readable open energy data portals so the schema and licensing are stable across refreshes.
A declared coordinate frame on every source. Ingestion does not guess. A source with crs is None is rejected, not defaulted to EPSG:4326 — an assumed frame is the most expensive kind of silent error. Ingestion records the native EPSG so the CRS alignment stage has an explicit source to transform from.
A target schema expressed as a pandera DataFrameSchema (the row-level contract) plus a pydantic model for any non-tabular configuration (endpoints, credentials, batch policy). Attribute-level rules such as valid voltage_kv classes belong here and are shared with downstream network attribute validation.
Library versions: geopandas >= 0.14, shapely >= 2.0 (for to_wkb with the flavor argument), pyproj >= 3.4, pandera >= 0.18, pydantic >= 2.5, pyarrow >= 14, and fsspec with the relevant backend (s3fs, gcsfs, or adlfs). The Arrow-native streaming path in the performance section additionally assumes geopandas >= 1.0 for zero-copy geometry from a pyarrow.Table.

Geometry validity is checked and recorded at ingestion but repaired downstream. Keeping make_valid out of the boundary is deliberate: an ingestion layer that quietly rewrites geometry hides exactly the data-quality signal that the spatial data quality & validation stage exists to surface and audit.

Core Implementation: A Deterministic Ingestion Function

The happy path is three moves: read whatever format the source is in, validate it against the contract, and stamp each surviving row with a deterministic key and lineage. The contract itself is the anchor — define it once and every source is measured against the same shape.

python

import pandera as pa
from pandera import Column, Check, DataFrameSchema

# The single row-level contract every source is coerced to.
ENERGY_ASSET_CONTRACT = DataFrameSchema(
    {
        "asset_id":    Column(str, nullable=False),
        "capacity_mw": Column(float, Check.in_range(0.0, 5_000.0), nullable=False),
        "voltage_kv":  Column(
            float,
            Check.isin([69.0, 115.0, 138.0, 161.0, 230.0, 345.0, 500.0, 765.0]),
            nullable=True,  # generation sites have no voltage class until interconnected
        ),
        "source_epsg": Column(int, Check.gt(1024), nullable=False),
        "geometry":    Column("geometry", nullable=False),
    },
    strict="filter",  # silently drop columns outside the contract, keep declared ones
    coerce=True,      # cast to the declared dtype or fail loudly
)

The deterministic asset key is what makes loading idempotent. It is a content hash over the stable identity of a feature — its id, its declared frame, and its geometry — so the same asset produces the same key on every run and a replayed batch collides with itself instead of duplicating.

python

import hashlib
from shapely import to_wkb


def deterministic_asset_key(asset_id: str, geom, source_epsg: int) -> str:
    """
    Content-addressed key over a feature's stable identity. Two ingests of the
    same asset yield the same key, so replay is an upsert, not an append.
    """
    h = hashlib.sha256()
    h.update(asset_id.encode("utf-8"))
    h.update(str(source_epsg).encode("utf-8"))
    # ISO WKB is canonical and endianness-stable across platforms.
    h.update(to_wkb(geom, output_dimension=2, flavor="iso"))
    return h.hexdigest()

A 256-bit digest is not superstition. With $n$ distinct assets and a $b$ -bit hash, the probability of any key collision follows the birthday bound

p (collision) \approx 1 - e^{- n^{2} / 2^{b + 1}} .

For a ten-million-asset national register against $b = 256$ , $p$ is on the order of $1 0^{- 63}$ — collisions are not a failure mode worth engineering around, which is exactly why a content hash is safe to use as a primary key.

The reader dispatches on format so a single call site handles the whole heterogeneous set, and the ingest function stamps lineage without ever mutating coordinates:

python

from pathlib import PurePosixPath
import fsspec
import geopandas as gpd
import pandas as pd


def read_source(uri: str, *, storage_options: dict | None = None) -> gpd.GeoDataFrame:
    """Format-dispatched reader for local paths and cloud object-store URIs."""
    suffix = PurePosixPath(uri.split("?", 1)[0]).suffix.lower()
    if suffix in {".parquet", ".geoparquet"}:
        return gpd.read_parquet(uri, storage_options=storage_options)
    if suffix in {".geojson", ".json", ".gpkg", ".shp"}:
        # Stream the bytes through fsspec so s3://, gs:// and az:// all work.
        with fsspec.open(uri, "rb", **(storage_options or {})) as handle:
            return gpd.read_file(handle)
    raise ValueError(f"Unsupported source format: {suffix!r}")


def ingest_source(
    uri: str,
    *,
    contract: DataFrameSchema = ENERGY_ASSET_CONTRACT,
    storage_options: dict | None = None,
    batch_id: str | None = None,
) -> gpd.GeoDataFrame:
    """
    Read, validate at the boundary, and emit the canonical contract frame.
    Records the native CRS as an attribute; does NOT reproject or repair.
    """
    raw = read_source(uri, storage_options=storage_options)
    if raw.crs is None:
        raise ValueError(f"{uri}: no declared CRS; ingestion refuses undated coordinates.")

    raw = raw.copy()
    raw["source_epsg"] = int(raw.crs.to_epsg())

    # lazy=True collects EVERY violation into one report instead of aborting on the first.
    validated = contract.validate(raw, lazy=True)

    validated["asset_key"] = [
        deterministic_asset_key(a, g, e)
        for a, g, e in zip(validated["asset_id"], validated.geometry, validated["source_epsg"])
    ]
    validated["batch_id"] = batch_id or f"auto-{pd.Timestamp.utcnow():%Y%m%dT%H%M%S}"
    validated["ingested_at"] = pd.Timestamp.utcnow().isoformat()
    return validated.set_geometry("geometry")

Every accepted row now carries source_epsg, asset_key, batch_id, and ingested_at. Those four columns are the entire contract the rest of the pipeline needs: the frame to transform from, the identity to deduplicate on, and the lineage to audit against.

Error Handling & Partial-Write Safety

Each of the failure modes named earlier gets a concrete guard.

Schema violations must be legible, not silent. Calling validate(..., lazy=True) collects every failing row and column into a single SchemaErrors exception rather than dying on the first bad cell. A production ingest catches it, writes the offending rows to a dead-letter store keyed by batch_id, and continues — quarantining is auditable, discarding is not.

python

from pandera.errors import SchemaErrors


def ingest_with_quarantine(uri, *, contract=ENERGY_ASSET_CONTRACT, storage_options=None):
    try:
        return ingest_source(uri, contract=contract, storage_options=storage_options), None
    except SchemaErrors as exc:
        # exc.failure_cases is a DataFrame: column, check, failure_case, index.
        failures = exc.failure_cases.assign(source_uri=uri)
        return None, failures  # caller persists `failures` to the dead-letter store

Partial writes must be atomic. Never serialise in place over a live target. Write to a unique staging key, then rename — on a POSIX filesystem the rename is atomic, and on object stores it is atomic at the single-key level, so a reader never observes a half-written file. Combine that with the deterministic key to make the whole load idempotent: concatenate, keep the newest row per key, and swap.

python

import uuid


def idempotent_upsert(new_gdf, target_uri, *, key="asset_key", storage_options=None):
    """
    Merge a validated batch into a GeoParquet target with replay safety:
    dedupe on the deterministic key, write to staging, then atomic-rename.
    """
    fs, _ = fsspec.core.url_to_fs(target_uri, **(storage_options or {}))
    staging_uri = f"{target_uri}.staging-{uuid.uuid4().hex}"

    if fs.exists(target_uri):
        existing = gpd.read_parquet(target_uri, storage_options=storage_options)
        merged = pd.concat([existing, new_gdf], ignore_index=True)
    else:
        merged = new_gdf

    # Latest ingest wins; a replayed batch collapses onto its own keys.
    merged = (
        merged.sort_values("ingested_at")
              .drop_duplicates(subset=key, keep="last")
              .reset_index(drop=True)
    )
    merged = gpd.GeoDataFrame(merged, geometry="geometry", crs=new_gdf.crs)
    merged.to_parquet(staging_uri, storage_options=storage_options)
    fs.mv(staging_uri, target_uri)  # atomic swap; readers see old or new, never partial
    return len(merged)

Transient I/O failures must retry without duplicating. Because the load is idempotent, a retry is always safe — re-running a batch that half-committed cannot double-count. Wrap the network-touching calls in bounded exponential backoff and let the deterministic key absorb any overlap between the failed attempt and the retry. The one caveat worth stating plainly: fs.mv is atomic per key, not transactional across a partitioned dataset, so a multi-file write should stage every part first and rename them in a fixed order, or write a _SUCCESS marker last and treat its absence as “ignore this batch.”

Performance & Scalability: Streaming GeoParquet Partitions

The scenario that breaks a naive ingest is a national GeoParquet dataset — tens of gigabytes of interconnection-queue points partitioned by state — sitting in s3://. Downloading it to local disk to read_parquet the whole thing wastes the format’s entire advantage. GeoParquet is columnar and row-grouped, so the right pattern reads only the columns and partitions a job needs, streams them straight from object storage, and never stages a byte to disk.

The generator below walks a Hive-partitioned dataset fragment by fragment, pushes the partition filter down into Arrow so unmatched files are never opened, and validates each partition against the same contract as it streams. Peak memory is bounded by the largest single row group, not the dataset — the full walkthrough of tuning row-group size, predicate push-down, and geometry-column decoding lives in streaming GeoParquet from cloud object storage with GeoPandas.

python

import pyarrow.dataset as ds
import pyarrow.compute as pc
from typing import Iterable, Iterator


def stream_geoparquet_partitions(
    root_uri: str,
    *,
    states: Iterable[str] | None = None,
    columns: list[str] | None = None,
    contract: DataFrameSchema = ENERGY_ASSET_CONTRACT,
    storage_options: dict | None = None,
) -> Iterator[gpd.GeoDataFrame]:
    """
    Stream and validate GeoParquet partitions directly from object storage.
    Filter push-down skips unmatched files; memory scales with one row group.
    """
    fs, root = fsspec.core.url_to_fs(root_uri, **(storage_options or {}))
    dataset = ds.dataset(root, filesystem=fs, format="parquet", partitioning="hive")

    predicate = pc.field("state").isin(list(states)) if states is not None else None

    for fragment in dataset.get_fragments(filter=predicate):
        table = fragment.to_table(filter=predicate, columns=columns)
        if table.num_rows == 0:
            continue  # partition pruned by the filter — never materialised
        # geopandas >= 1.0 reads the GeoParquet geometry metadata zero-copy.
        partition_gdf = gpd.GeoDataFrame.from_arrow(table)
        partition_gdf["source_epsg"] = int(partition_gdf.crs.to_epsg())
        yield contract.validate(partition_gdf, lazy=True)

Additional scaling levers for continental ingests:

Push filters down, do not filter after. Passing the predicate to to_table lets Arrow skip whole files and row groups using their statistics. Reading everything and filtering the resulting frame moves the same bytes you were trying to avoid.
Project columns early. A capacity screen rarely needs every attribute. Passing columns=["asset_id", "capacity_mw", "geometry", "state"] reads a fraction of each row group off the wire.
Size row groups to the reader. Row groups that are too large defeat memory bounding; too small and per-group overhead dominates. Target groups that decode to a few hundred megabytes for a streaming worker.
Parallelise across fragments, bound the concurrency. Fragments are independent, so a pool of workers each handling a partition scales linearly — but cap the pool so a wide dataset does not open thousands of sockets against the object store at once.
Never round-trip through disk. The whole point of fsspec + Arrow is that bytes flow object-store → Arrow → GeoDataFrame. A download-then-read step reintroduces the I/O and the local-capacity limit the streaming path exists to remove.

Validation & Audit Trail

An ingested batch is only trustworthy once it can be described and re-verified. The final move emits a manifest — a compact, structured record of what entered the pipeline — and asserts the invariants the contract promised. A duplicate key inside a single batch, for instance, means an upstream asset_id is not actually unique, and that must fail loudly at ingestion rather than deduplicate silently and understate the asset count.

python

def build_ingest_manifest(validated_gdf, *, source_uri: str, schema_version: str) -> dict:
    """
    Structured, queryable lineage for one ingested batch. Raises if the
    deterministic-key invariant was violated upstream.
    """
    epsg_counts = validated_gdf["source_epsg"].value_counts().to_dict()
    manifest = {
        "source_uri": source_uri,
        "schema_version": schema_version,
        "batch_id": validated_gdf["batch_id"].iloc[0] if len(validated_gdf) else None,
        "record_count": int(len(validated_gdf)),
        "distinct_asset_keys": int(validated_gdf["asset_key"].nunique()),
        "source_epsg_distribution": {int(k): int(v) for k, v in epsg_counts.items()},
        "capacity_mw_total": float(validated_gdf["capacity_mw"].sum()),
        "ingested_at": pd.Timestamp.utcnow().isoformat(),
    }
    # Invariant: within one batch, the deterministic key is unique.
    if manifest["distinct_asset_keys"] != manifest["record_count"]:
        raise ValueError(
            "Duplicate asset_key within a batch — upstream asset_id is non-unique; "
            "resolve at the source before loading."
        )
    return manifest

The source_epsg_distribution is more than bookkeeping: a batch that suddenly shows a second EPSG code is a strong early signal that a provider changed their export projection, and it is the exact fact the downstream CRS alignment stage needs to reconcile. Emit the manifest as JSON to a log aggregator and wire two assertions into continuous integration — the deterministic-key invariant above, and a capacity_mw_total that stays within a tolerance of the previous run’s total for the same source. A scheduled ingest of a known fixture that checks record count, CRS distribution, and capacity total against stored values will catch a schema regression before it reaches a permitting or interconnection deliverable.

At production scale, treat the whole sequence as a deterministic, auditable pipeline rather than a load script: validate once at the boundary, key every record content-addressably so replay is safe, stream partitions instead of staging them, and carry the manifest forward as the lineage that lets a reviewer reproduce the batch. That discipline is what lets energy and GIS teams fold a dozen incompatible source formats into one predictable contract — and hand the projection, topology, and network-attribute stages a frame they never have to second-guess.

Core Energy-GIS Data & Spatial Fundamentals — the six-stage pipeline this ingestion boundary is the first stage of.
Streaming GeoParquet from Cloud Object Storage with GeoPandas — the full walkthrough of partition streaming, predicate push-down, and row-group tuning.
Coordinate Reference Systems for Energy Projects — the alignment stage that consumes the source_epsg ingestion records.
Spatial Data Quality & Validation — the geometry repair and topology rules ingestion deliberately defers to.
Open Energy Data Portals — versioned, machine-readable sources whose stable schemas make the ingestion contract enforceable.
Network Attribute Validation — the downstream stage that shares the voltage_kv and capacity attribute rules defined in the contract.

Geospatial Data Ingestion Pipelines #

The Ingestion Contract: Validate at the Boundary #

Prerequisites & Data Requirements #

Core Implementation: A Deterministic Ingestion Function #

Error Handling & Partial-Write Safety #

Performance & Scalability: Streaming GeoParquet Partitions #

Validation & Audit Trail #

Related #

Explore this section

Geospatial Data Ingestion Pipelines

The Ingestion Contract: Validate at the Boundary

Prerequisites & Data Requirements

Core Implementation: A Deterministic Ingestion Function

Error Handling & Partial-Write Safety

Performance & Scalability: Streaming GeoParquet Partitions

Validation & Audit Trail

Related