🔧 Fix Batch processing Testing, CI & Docker Updated

Batch Job Gives Different Results Each Run: How to Fix It

Problem statement

Nothing changed. The output did.

$ python -m pipeline run && sha256sum output/summary.gpkg
9f2c1a...  output/summary.gpkg

$ python -m pipeline run && sha256sum output/summary.gpkg
4d81e7...  output/summary.gpkg        # same input, same code, different bytes

Sometimes it is only the bytes — a timestamp in the file header, harmless. Sometimes it is the data:

run 1:  12,180 rows, total area 4,813,880.2 m²
run 2:  12,180 rows, total area 4,813,880.2 m²
run 3:  12,177 rows, total area 4,811,204.8 m²      ← three parcels went missing

A job that produces different answers from the same input cannot be tested, cannot be verified against yesterday, and cannot be trusted when a number looks wrong — because "run it again" is no longer a way to check anything.

Non-determinism in GIS pipelines has a small number of causes, and almost all of them are an unordered collection being consumed as if it were ordered.

Quick answer

Find the source by bisecting on the output, then remove it:

import hashlib, geopandas as gpd

def fingerprint(gdf, *, sort_by=None):
    """A hash of the DATA, insensitive to row order and file encoding."""
    g = gdf.copy()
    if sort_by:
        g = g.sort_values(sort_by)
    g = g.sort_index(axis=1)                                   # column order
    wkb = g.geometry.to_wkb()
    attrs = g.drop(columns=g.geometry.name).astype(str)
    payload = (attrs.agg("|".join, axis=1) + "|" + wkb.map(bytes.hex)).sort_values()
    return hashlib.blake2b("\n".join(payload).encode(), digest_size=16).hexdigest()

a = fingerprint(run_pipeline(), sort_by="parcel_id")
b = fingerprint(run_pipeline(), sort_by="parcel_id")
print("deterministic" if a == b else "NON-DETERMINISTIC")
Cause Symptom Fix
unsorted glob / rglob files processed in different order sorted()
set or dict iteration groups or keys in different order sort before iterating
drop_duplicates without sort a different row survives each run sort first, then drop
parallel results collected as they finish row order varies sort the results, or key them
dissolve / groupby on unordered input aggregation of first differs sort before grouping
floating-point summation order totals differ in the last digits sort, or use a stable reducer
a timestamp written into the output bytes differ, data does not strip or fix it
Path.iterdir() filesystem order, varies by machine sorted() 
# the four lines that fix most of it
for path in sorted(src.rglob("*.gpkg")):            # not rglob alone
    ...
gdf = gdf.sort_values("parcel_id").reset_index(drop=True)
gdf = gdf.loc[~gdf["parcel_id"].duplicated(keep="first")]     # after sorting
results = sorted(pool_results, key=lambda r: r["item"])

Ordered versus unordered

Grid of collection types showing which guarantee order and which do not.
Anything in the right-hand column produces a different answer on a different machine.

Step-by-step solution

Triage rows pairing each non-determinism symptom with its cause and fix.
Bisect on the fingerprint. The step where two runs diverge is the step with the bug.

1. Separate "different bytes" from "different data" 

# byte comparison — fails on harmless header timestamps
sha256sum output/a.gpkg

# data comparison — only fails when the data is genuinely different
fingerprint(gpd.read_file("output/a.gpkg"), sort_by="parcel_id")

A GeoPackage stores a creation timestamp, SQLite reorders pages, and shapefile writes a .shp header with its own metadata. All of that changes the bytes without changing a single feature. Chasing byte-level reproducibility first wastes a day; the data fingerprint answers the question that matters.

If byte-identical output is genuinely required — for a checksum-based cache, or a regulated delivery — the practical route is to normalise on read rather than to fight the writers:

def canonical_bytes(path, sort_by):
    gdf = gpd.read_file(path).sort_values(sort_by).reset_index(drop=True)
    return gdf.to_json(sort_keys=True).encode()      # deterministic serialisation

2. Bisect: fingerprint after every step 

def traced(steps, gdf, sort_by):
    prints = []
    for name, fn in steps:
        gdf = fn(gdf)
        prints.append((name, fingerprint(gdf, sort_by=sort_by), len(gdf)))
    return gdf, prints

_, a = traced(STEPS, load(), "parcel_id")
_, b = traced(STEPS, load(), "parcel_id")

for (name, ha, na), (_, hb, nb) in zip(a, b):
    mark = "  ← diverges here" if ha != hb else ""
    print(f"{name:32s} {ha[:12]}  {na:>7,}{mark}")
read                             3f8a1c9e2d41   12,400
drop null geometry               3f8a1c9e2d41   12,394
make_valid                       7b2e4f01aa93   12,394
deduplicate                      c1d9…/e04b…    12,177  ← diverges here
dissolve by ward                 …                  32

One line of output, one line of code to look at. This beats reading the whole pipeline hoping to spot the problem.

3. Sort every discovery, without exception 

# these all return filesystem order, which varies by OS, filesystem and inode reuse
glob.glob("data/*.gpkg")
Path("data").rglob("*.gpkg")
Path("data").iterdir()
os.listdir("data")

# always
sorted(Path("data").rglob("*.gpkg"))

This is the single most common cause. Filesystem order is not random — it is stable on one machine and different on another, which is why the job is deterministic locally and not in CI. See python glob is not finding all my shapefiles for the related discovery trap.

It matters even when files are processed independently, because anything that appends to a shared list, writes to a shared log, or assigns sequential ids inherits the order.

4. Sort before any operation that picks a winner 

# non-deterministic: which duplicate survives depends on arrival order
gdf = gdf.drop_duplicates(subset="parcel_id")

# deterministic: the rule is explicit
gdf = (gdf.sort_values(["parcel_id", "surveyed"], ascending=[True, False])
          .drop_duplicates(subset="parcel_id", keep="first"))

The same applies to every "pick one" operation:

gdf.dissolve(by="ward", aggfunc="first")          # first of WHAT order?
gdf.groupby("ward").first()
gpd.sjoin(...).loc[~index.duplicated(keep="first")]

first is only meaningful after a sort. Without one it means "whatever pandas encountered first", which depends on how the frame was built.

gdf = gdf.sort_values(["ward", "parcel_id"])
dissolved = gdf.dissolve(by="ward", aggfunc="first")     # now reproducible

5. Order parallel results before using them 

from concurrent.futures import ProcessPoolExecutor, as_completed

# as_completed yields in completion order — genuinely non-deterministic
results = []
for fut in as_completed(futures):
    results.append(fut.result())

# fix: sort by a stable key from the item, not by arrival
results.sort(key=lambda r: r["item"])

# or use map, which preserves input order
with ProcessPoolExecutor(8) as pool:
    results = list(pool.map(process, sorted(items)))

as_completed is the right tool for progress reporting and the wrong one for building an ordered result. Use it, then sort — or use map when order matters more than early feedback.

6. Deal with floating-point summation order 

import numpy as np
values = np.random.default_rng(0).random(1_000_000) * 1e6
a = values.sum()
b = values[::-1].sum()
a == b            # False — differs in the last few bits

Floating-point addition is not associative, so a total genuinely depends on the order of the terms. In a parallel reduction the order varies with worker timing, and the last two digits of an area total move between runs.

This is usually harmless and occasionally not — a != comparison against a stored total, or a hash of the numbers, will flip.

# make it stable: sort before summing
total = float(np.sort(values).sum())

# or accept it, and compare with a tolerance instead of equality
assert abs(total - expected) < 1e-6 * abs(expected)

Rounding to a sensible precision at the boundary is usually the cleanest answer — areas to the square metre, lengths to the millimetre. The extra digits were never real.

7. Strip timestamps and machine identity from outputs 

metadata = {
    "generated_at": datetime.now().isoformat(),      # differs every run
    "host": socket.gethostname(),                    # differs every machine
    "run_id": uuid.uuid4().hex,                      # differs always
    "source_hash": content_hash(src),                # ← the useful one
    "recipe": RECIPE,
}

Run metadata is worth keeping — see how to record run metadata and data lineage. Keep it in a sidecar file rather than inside the data, so the data can be compared and the provenance is still recorded.

output/
├── summary.gpkg          ← deterministic; compare this
└── summary.run.json      ← timestamps, host, run id

Code examples

Example 1: a determinism test 

import pytest

def test_pipeline_is_deterministic(tmp_path, sample_input):
    a = run_pipeline(sample_input, tmp_path / "a")
    b = run_pipeline(sample_input, tmp_path / "b")
    assert fingerprint(a, sort_by="parcel_id") == fingerprint(b, sort_by="parcel_id")

def test_pipeline_is_order_independent(tmp_path, sample_input, monkeypatch):
    """Same files, reversed discovery order — the output must not change."""
    a = run_pipeline(sample_input, tmp_path / "a")
    monkeypatch.setattr("pipeline.discover", lambda s: list(reversed(discover(s))))
    b = run_pipeline(sample_input, tmp_path / "b")
    assert fingerprint(a, sort_by="parcel_id") == fingerprint(b, sort_by="parcel_id")

@pytest.mark.parametrize("workers", [1, 4])
def test_worker_count_does_not_change_output(tmp_path, sample_input, workers):
    out = run_pipeline(sample_input, tmp_path / str(workers), workers=workers)
    assert fingerprint(out, sort_by="parcel_id") == GOLDEN_FINGERPRINT

The second test is the valuable one. Running twice on one machine often passes by luck, because the filesystem returns the same order both times. Reversing discovery order deliberately is what proves the pipeline does not depend on it.

The third catches the parallel-collection bug, which is invisible at workers=1.

Example 2: a determinism guard for the pipeline itself 

def assert_deterministic_inputs(paths):
    """Refuse to run on a work list that is not stably ordered."""
    assert paths == sorted(paths), (
        "the work list is not sorted — output will depend on filesystem order"
    )

def assert_sorted_before(gdf, cols, operation):
    ordered = gdf[cols].reset_index(drop=True)
    assert ordered.equals(ordered.sort_values(cols).reset_index(drop=True)), (
        f"{operation} needs {cols} sorted first, or its result is arbitrary"
    )

Cheap assertions at the point of risk, rather than a test that runs the whole pipeline twice.

Example 3: the fingerprint, as a run-over-run check 

def compare_runs(current: gpd.GeoDataFrame, previous_path: Path, sort_by: str):
    if not previous_path.exists():
        return {"status": "no baseline"}
    previous = gpd.read_file(previous_path)
    fc, fp = fingerprint(current, sort_by=sort_by), fingerprint(previous, sort_by=sort_by)
    if fc == fp:
        return {"status": "identical"}
    return {
        "status": "changed",
        "rows": (len(previous), len(current)),
        "new_ids": sorted(set(current[sort_by]) - set(previous[sort_by]))[:10],
        "gone_ids": sorted(set(previous[sort_by]) - set(current[sort_by]))[:10],
    }

Once the job is deterministic, this becomes genuinely useful: any difference between last night and tonight is a real data change, and the job can report exactly what moved. That is impossible while the output is noisy.

Explanation

Checklist of capabilities that a deterministic pipeline unlocks.
Every one of these is impossible while the output varies for no reason.

Almost all non-determinism in a GIS pipeline traces to the same root: a collection that guarantees no order being consumed as if it did. Filesystem listings, Python sets, dict iteration in older versions, as_completed, thread scheduling — all of them are documented as unordered, and all of them appear ordered in casual testing because they are stable on one machine. That stability is what makes the bug so slippery: it reproduces perfectly until it runs somewhere else.

The consequences propagate. An unordered file list changes which row drop_duplicates keeps, which changes which geometry survives, which changes an area total, which changes a percentage in a report. A difference that started as inode ordering ends up as a number somebody has to explain.

Floating-point summation is the one genuinely unavoidable case. Addition is not associative in IEEE 754, so a parallel reduction over the same values in a different grouping produces a slightly different total — correctly. The fix is not to eliminate it but to stop depending on exact equality: round at the boundary to a precision the data actually supports, and compare with tolerances.

The payoff for fixing all of this is larger than "the hashes match". A deterministic pipeline can be tested (a golden result stays golden), diffed (any change between runs is a real change), cached (the incremental job can trust that unchanged input means unchanged output), and debugged (re-running reproduces the problem). None of those are available while the output moves on its own, which is why this is worth an afternoon even when the variation looks harmless.

Edge cases or notes

  • Python dicts preserve insertion order since 3.7, so dict iteration is deterministic given deterministic insertion. Sets never are.
  • PYTHONHASHSEED affects set iteration order across processes; it is randomised by default. Never rely on set order, even within one machine.
  • geopandas.sjoin result order depends on the index implementation, which can change between versions. Sort after joining.
  • dissolve sorts by the group key by default — that part is deterministic; aggfunc="first" over unsorted rows is not.
  • Multiprocessing on fork versus spawn changes module import order, which can change anything relying on import side effects.
  • GeoPackage stores a last_change timestamp in gpkg_contents, so byte-identical output requires rewriting it.
  • to_file on shapefile writes the current date into the .dbf header.
  • Sorting a large frame is not free — sort once at the boundary rather than before every operation.

FAQ

The bytes differ but the data looks the same. Does it matter?

Usually not — GeoPackage writes a creation timestamp and SQLite reorders pages. Compare a fingerprint of the data rather than the file, unless byte-identical output is a requirement.

Why is it deterministic on my machine and not in CI?

Filesystem ordering is stable per machine and differs between them. An unsorted glob reproduces perfectly locally and changes in CI.

Do I have to sort everything?

Sort discovery, and sort before any operation that picks a winner — drop_duplicates, first, deduplicating a join. Elsewhere it is unnecessary overhead.

Why does my area total change in the last two digits?

Floating-point addition is not associative, so a parallel reduction sums in a different order. Round at the boundary, and compare with a tolerance rather than for equality.

Does as_completed cause this?

Yes, if you build an ordered result from it. Use it for progress, then sort the results by a stable key — or use pool.map, which preserves input order.

How do I prove the pipeline is deterministic?

Run it twice with discovery order reversed, and at two different worker counts. Running it twice unchanged often passes by luck.

Is a golden test worth it?

Once the pipeline is deterministic, yes — it becomes the cheapest regression test available. Before then it fails constantly and gets deleted.