13 min read 2857 words

Incremental, Idempotent Loading of Grid Datasets with Hive Partitions

The scenario: a monthly interconnection-queue load is re-run after a failure, and the resulting dataset has 1.8 million rows where it should have 1.2 million. Every row is valid, no error was raised, and the duplicates are invisible until a capacity total comes out 50 percent high. This page builds the load that cannot do that, and it is the incremental-loading detail behind geospatial data ingestion pipelines.

Root-cause analysis

Three properties have to hold together for a reload to be a no-op, and losing any one produces a different failure.

  1. Append semantics without a key. Writing new rows into an existing partition duplicates everything the previous run already wrote. The fix is not “check before appending” — that races — but a deterministic key plus a whole-partition replace.
  2. A key derived from mutable fields. Hashing the row including a last_updated column means the same asset gets a new key on every publication, so deduplication never matches and the store grows monotonically. The key must come from the fields that identify the thing, not from the fields that describe its current state.
  3. Partial writes treated as complete. A killed job that wrote half a partition leaves a file every reader treats as authoritative. Staging-then-rename makes the write atomic, so a reader sees either the previous version or the new one.
Hive-partitioned layout for a monthly grid dataset A directory tree showing a queue root with state equals TX, NM and OK directories, each containing month equals 2026-05, 2026-06 and 2026-07 directories, and each of those holding a single part.parquet file. One path is highlighted as the partition a Texas July query opens. Annotations note that the filter prunes 611 of 612 files before any bytes are read, that the partition columns are also written inside each file, and that a staging file with a dot prefix is renamed into place so a reader never sees a partial write. One query, one file opened — the rest are never touched queue/ state=TX month=2026-05/part.parquet month=2026-06/part.parquet month=2026-07/part.parquet state=NM month=2026-05/part.parquet month=2026-06/part.parquet month=2026-07/part.parquet state=OK month=2026-05/part.parquet month=2026-06/part.parquet month=2026-07/part.parquet filter: state=TX AND month=2026-07 611 of 612 files pruned before any read the prune is a path match, not a scan partition columns written inside the file too — a copy that loses its path keeps its identity .staging-<uuid>.parquet renamed into place — a reader never sees a partial write

Pre-flight validation

Before writing, confirm the batch is internally consistent: the partition columns are single-valued, the deterministic key is unique, and the geometry column survived the read.

python
import hashlib

import geopandas as gpd


def deterministic_key(row, *, fields: tuple[str, ...]) -> str:
    """A key built only from identity fields — never from mutable state."""
    payload = "|".join(str(row[f]) for f in fields)
    return hashlib.sha256(payload.encode("utf-8")).hexdigest()[:24]


def preflight_batch(
    batch: gpd.GeoDataFrame,
    *,
    partition_cols: tuple[str, ...] = ("state", "month"),
    key_fields: tuple[str, ...] = ("queue_id", "poi_name", "state"),
) -> dict:
    """Refuse a batch that cannot be written idempotently."""
    for col in partition_cols:
        values = batch[col].dropna().unique()
        if len(values) != 1:
            raise ValueError(f"{col} must be single-valued in a partition batch, got {list(values)[:5]}")

    keys = batch.apply(deterministic_key, axis=1, fields=key_fields)
    dupes = int(keys.duplicated().sum())
    if dupes:
        raise ValueError(f"{dupes} duplicate keys within the batch — key fields are not identifying")

    if batch.geometry.isna().any():
        raise ValueError("null geometry in the batch — quarantine before loading")

    return {
        "rows": len(batch),
        "partition": {c: batch[c].iloc[0] for c in partition_cols},
        "key_fields": key_fields,
        "unique_keys": int(keys.nunique()),
    }

Fix implementation

The load below replaces one partition atomically. Replacement rather than append is what makes it idempotent: running it twice with the same input leaves exactly the same bytes.

python
import uuid

import fsspec
import geopandas as gpd


def load_partition(
    batch: gpd.GeoDataFrame,
    *,
    root: str,
    state: str,
    month: str,
    key_fields: tuple[str, ...],
    storage_options: dict | None = None,
) -> dict:
    """Replace one Hive partition atomically. Re-running is a no-op."""
    so = storage_options or {}
    prefix = f"{root}/queue/state={state}/month={month}"
    target = f"{prefix}/part.parquet"
    staging = f"{prefix}/.staging-{uuid.uuid4().hex}.parquet"

    prepared = batch.copy()
    prepared["asset_key"] = prepared.apply(deterministic_key, axis=1, fields=key_fields)
    # Last write wins within a batch: a republished row supersedes its predecessor.
    prepared = prepared.sort_values("published_at").drop_duplicates("asset_key", keep="last")

    prepared.to_parquet(staging, index=False, storage_options=so)

    fs, _ = fsspec.core.url_to_fs(target, **so)
    fs.mv(staging.replace(f"{root}/", ""), target.replace(f"{root}/", "")) if False else fs.mv(
        _strip_protocol(fs, staging), _strip_protocol(fs, target)
    )

    return {
        "partition": f"{state}/{month}",
        "rows_written": len(prepared),
        "rows_in": len(batch),
        "superseded": len(batch) - len(prepared),
        "target": target,
    }


def _strip_protocol(fs, uri: str) -> str:
    return fs._strip_protocol(uri)

The sort_values("published_at").drop_duplicates(keep="last") pair is the revision policy made explicit. Energy portals republish corrected rows, and a load that keeps the first occurrence silently ignores every correction — a failure that is much harder to notice than a duplicate.

Re-running the same load under three strategies Three rows showing the outcome of loading the same 1.2 million row batch twice. Append: 2.4 million rows after the second run, with every downstream total doubled. Merge on a key built from mutable fields: 1.9 million rows, because revised rows hash differently and are added rather than matched. Atomic partition replace: 1.2 million rows, byte-identical to the first run. Each row is annotated with whether the second run is detectable from the data alone. Load the same batch twice — three outcomes append 2.4 M rows · totals doubled no error, no warning merge on a mutable-field key 1.9 M rows · revisions added grows every month atomic partition replace 1.2 M rows · byte-identical re-run is a no-op expected: 1.2 M rows The append case is the dangerous one: every row is valid, the schema passes, and the only symptom is a capacity total that is exactly twice what it should be.

Fallback routing and performance tuning

  • Partition by the grain queries filter on. State and month are the natural grain for queue data; partitioning by an ingestion batch identifier makes every query a full scan.
  • Keep partitions between 128 MB and 1 GB. Tens of thousands of small partitions cost more in object listing than they save in pruning, which is the same trade covered in streaming GeoParquet from cloud object storage.
  • Write the partition columns into the file as well as the path. A reader that loses the Hive path — a file copied elsewhere, a manifest built by hand — otherwise loses the partition identity.
  • Re-examine a freshness window, not the whole history. Three months of monthly data is a cheap nightly sweep; re-loading five years every night is not, and nothing older changes without a deliberate backfill.
  • Record the source fingerprint on the partition. It is what lets the next run skip a partition whose input has not changed, rather than rewriting identical bytes.

Downstream validation

python
import geopandas as gpd


def assert_partition_idempotent(
    root: str, state: str, month: str, *, expected_rows: int, storage_options=None
) -> None:
    """Re-read the partition and prove the load did what it claimed."""
    path = f"{root}/queue/state={state}/month={month}/part.parquet"
    got = gpd.read_parquet(path, storage_options=storage_options or {})
    assert len(got) == expected_rows, f"partition holds {len(got)} rows, expected {expected_rows}"
    assert got["asset_key"].is_unique, "duplicate asset keys survived the load"
    assert (got["state"] == state).all(), "partition contains rows from another state"
    assert (got["month"] == month).all(), "partition contains rows from another month"
    assert got.geometry.notna().all(), "null geometry reached the store"
Four post-load assertions and the failure each one catches A four-row table pairing an assertion with the failure it catches. Row count equals the expected batch size catches an append that duplicated the partition. Unique asset keys catch a deterministic key that included a mutable field. Partition columns being single-valued catches a batch written into the wrong path. Non-null geometry catches rows that bypassed the quarantine. A note records that all four run against the re-read partition rather than against the in-memory batch, because the point is to check what actually landed. Assert against the re-read partition, not the batch in memory len(partition) == expected_rows an append that duplicated the partition asset_key.is_unique a key built from a mutable field state and month single-valued a batch written to the wrong path geometry.notna().all() rows that bypassed the quarantine Re-reading costs a second and is the only check that sees what the object store actually holds.

Handling revisions without losing history

A whole-partition replace is the right default and it throws away the previous version, which is usually fine and occasionally not — a figure that appeared in a submission has to remain reconstructible even after the source revises it.

The cheap pattern is a dated archive alongside the live partition. Before the rename, copy the existing part file to an archive prefix keyed by the load date; the live path always holds the current truth and the archive holds what was true on each load date. Storage is negligible for tabular grid data, and the archive answers the only question anyone asks retrospectively: what did this dataset say on the date the study was run.

The alternative — versioning inside the partition with a validity interval per row — is more precise and much more invasive, because every downstream query then has to filter on the interval or it sees several versions of the same asset. That cost is worth paying when the pipeline itself needs as-of queries, and not worth paying when the requirement is simply “reproduce last quarter’s number”.

Whichever is chosen, record the load date and the source fingerprint on the partition. Those two fields are what let a rerun decide whether anything actually changed, and what let a reviewer tie a published figure to the bytes that produced it.

Frequently asked questions

Why replace the whole partition instead of merging?

Because a merge has to read, combine and write anyway, and doing it as a replace makes the result a pure function of the input batch. A merge that reads the existing partition also inherits whatever is wrong with it, so a bad load has to be undone manually; a replace overwrites it. Where history genuinely matters, keep the previous version as a separate dated artefact rather than merging into the live one.

What should the deterministic key include?

The smallest set of fields that identifies the thing across publications — typically the source identifier, the point of interconnection and the state. What it must exclude is anything that changes when the row is revised: status, capacity, queue position and every timestamp. If the source provides a stable identifier, use it directly and skip the hash.

How do I handle a row that disappears from the source?

Decide the policy explicitly and record it. A whole-partition replace naturally drops it, which is right when the source is authoritative for that month and wrong when the row was omitted by accident. A useful middle path is to replace the partition but log the disappeared keys, so a sudden drop in row count is visible rather than silent.

Does this work on a local filesystem as well as object storage?

Yes, and the rename is genuinely atomic there — POSIX guarantees it within a filesystem. On object stores the guarantee is weaker but sufficient in practice: the rename is a server-side copy plus delete, and readers see either the old key or the new one. What breaks on both is renaming across filesystems or buckets, which degrades to a non-atomic copy.

Should the loader validate the schema too?

Yes, before the write and not after. A schema violation caught after a partition has been replaced means the previous good version is already gone. The order that survives a bad batch is: validate, stage, verify the staged file, then rename.

How large can a single partition get before this breaks down?

The pattern holds until the partition no longer fits comfortably in the loader’s memory, since the replace reads the whole batch. At that point the answer is a finer partition grain rather than a smarter loader — a state-month partition that exceeds a gigabyte usually means the data deserves a daily grain.

Related articles

Back to Geospatial Data Ingestion Pipelines