How to Process Only the Files That Changed Since the Last Run
On this page
Problem statement
A nightly job reprocesses every file it is given. On a typical night almost none of them are different.
for path in sorted(Path("data").rglob("*.gpkg")):
gdf = gpd.read_file(path)
gdf = transform(gdf)
gdf.to_file(out / path.name, driver="GPKG")Every night it rewrites outputs identical to the ones they replace. On a large folder the overnight window fills up, the downstream sync sees every file as changed instead of the few that were, and adding a new region will not fit. The job on this page is small — HM Land Registry's Leeds titles, split into one file per ward — so the saving is seconds rather than hours, but the logic, and every number printed below, is real.
The fix is to make the job skip work it has already done — but "already done" has to mean this exact input, through this exact transformation, or the job will silently stop updating the moment a supplier edits a file in place.
Quick answer
Keep a manifest of what was processed, compare against it, and process only the difference:
import hashlib
import json
from pathlib import Path
STATE = Path("state/manifest.json")
VERSION = "2026.08.1" # bump when the transformation changes
def content_hash(path, chunk=1 << 20):
h = hashlib.blake2b(digest_size=16)
with open(path, "rb") as f:
while block := f.read(chunk):
h.update(block)
return h.hexdigest()
def load_state():
return json.loads(STATE.read_text()) if STATE.exists() else {}
def save_state(state):
STATE.parent.mkdir(parents=True, exist_ok=True)
tmp = STATE.with_suffix(".tmp")
tmp.write_text(json.dumps(state, indent=2, sort_keys=True))
tmp.replace(STATE) # atomic: never a half-written manifest
def changed_files(src: Path, state: dict, recipe: str):
for path in sorted(src.rglob("*.gpkg")):
prev = state.get(str(path))
if prev is None:
yield path, "new"
elif prev["recipe"] != recipe:
yield path, "recipe"
elif prev["size"] != path.stat().st_size:
yield path, "size"
elif prev["hash"] != content_hash(path):
yield path, "content"state = load_state()
work = list(changed_files(Path("data"), state, VERSION))
print(f"{len(work)} of {len(list(Path('data').rglob('*.gpkg')))} files need processing")
for path, reason in work:
result = process(path)
if result.ok:
state[str(path)] = {
"hash": content_hash(path), "size": path.stat().st_size,
"recipe": VERSION, "processed_at": now_iso(), "rows": result.rows,
}
save_state(state) # once, after the loopThe complete job is Example 1, and its runs are printed there.
What the manifest holds
What you get — from The data used on this page
33 ward files, 158 MB
Change detection in discovery, size before hash, a recipe id
1. Put change detection in discovery, not in the loop
# wrong — the check is buried, and a dry run still opens every file
for path in all_files:
if not needs_processing(path):
continue
process(path)
# right — discovery returns the work list, and can be inspected alone
work = plan(src, state, recipe)
print(f"{len(work)} items")
for item in work:
process(item)Keeping it in discovery means --dry-run is free, the plan can be printed and reviewed, and the count is known before any work starts. That is the discover/apply/report shape paying for itself.
2. Use size as a free pre-filter before hashing
if prev["size"] != path.stat().st_size:
return True # different size = certainly different content
return prev["hash"] != content_hash(path) # same size: must read to be sureA stat() is microseconds; hashing gigabytes is seconds. Many genuine changes alter the size, so the cheap check often resolves them without reading anything — but not all. On the third night below, deleting one title from a ward's GeoPackage left the file exactly the same size, because SQLite allocates in whole pages, and only the hash noticed. The size check is a shortcut for "certainly changed", never for "unchanged".
If even hashing every unchanged file is too slow — a network share, or terabytes of imagery — use mtime as the pre-filter and hash only what looks recent:
def maybe_changed(path, prev):
if prev["size"] != path.stat().st_size:
return True
if path.stat().st_mtime <= prev["mtime"]:
return False # not touched since we last saw it
return prev["hash"] != content_hash(path) # touched — confirm with a hashThis is fast and correct in the direction that matters: a file whose mtime went backwards is not something any normal tool produces.
3. Add a recipe id, so code changes invalidate outputs
import inspect
def recipe_id(config: dict, *functions) -> str:
h = hashlib.blake2b(digest_size=8)
h.update(json.dumps(config, sort_keys=True, default=str).encode())
for fn in functions:
h.update(inspect.getsource(fn).encode())
return h.hexdigest()
Without this, changing the target CRS from 27700 to 3857 produces a run that skips every file, reports success, and leaves every output in the old projection. Hashing the function source means the invalidation happens automatically — nobody has to remember to bump a version.
4. Guard against an accidental full rebuild
def guard(work, total, *, threshold=0.5, force=False):
if force or not total:
return work
if len(work) / total > threshold:
raise SystemExit(
f"{len(work):,}/{total:,} ({len(work)/total:.0%}) look changed — "
f"that is unusual. Check state/manifest.json and the recipe id, "
f"then re-run with --force if this really is a full rebuild."
)
return workA corrupted manifest, a recipe id that accidentally includes a timestamp, or a supplier re-exporting everything all produce the same symptom: every item to process. Two of those three are bugs. Failing loudly turns a wasted overnight window into a thirty-second check.
5. Write the output atomically, then record it
def process(path: Path, out_dir: Path):
gdf = gpd.read_file(path)
gdf = transform(gdf)
final = out_dir / path.name
tmp = final.with_suffix(".tmp.gpkg")
gdf.to_file(tmp, driver="GPKG")
tmp.replace(final) # atomic on the same filesystem
return Result(ok=True, rows=len(gdf))Two properties matter and they reinforce each other. Atomic output means a crash never leaves a truncated file that looks complete — which would then be skipped forever by an "output exists" check. Recording after means the manifest never claims work that did not finish.
for path, reason in work:
result = process(path, out_dir)
if result.ok:
state[str(path)] = fingerprint(path, RECIPE, result)
# a failure leaves no manifest entry, so the next run retries it
save_state(state)The failure path is the elegant part: a failed item simply has no manifest entry, so the next run picks it up again. Retry logic falls out of the design rather than being added.
6. Handle inputs that disappear
def prune(state, src: Path, out_dir: Path, *, delete_outputs=False):
gone = [k for k in state if not Path(k).exists()]
for key in gone:
entry = state.pop(key)
if delete_outputs and entry.get("output"):
Path(entry["output"]).unlink(missing_ok=True)
return goneWithout pruning, the manifest grows forever and the outputs of deleted inputs linger, so the output directory slowly stops matching the input. Whether to delete the outputs is a policy decision — for a mirror, yes; for an archive, no.
An incremental job over Leeds ward files, proved
The data used on this page
HM Land Registry's INSPIRE polygons for Leeds, split into one GeoPackage per electoral ward — 33 files in data/ — so the job has a real folder of inputs to be incremental about. Each title goes to the ward containing a point inside it. Land Registry data © Crown copyright and database right 2026, OGL v3; wards from the ONS.
import sys
from pathlib import Path
import geopandas as gpd
import pyogrio
import requests
reply = requests.Session().get("https://use-land-property-data.service.gov.uk/datasets/inspire/download/Leeds.zip",
timeout=600) # sets a cookie, then redirects to itself
reply.raise_for_status()
Path("leeds_inspire.zip").write_bytes(reply.content)
pyogrio.set_gdal_config_options({"GML_DOWNLOAD_SCHEMA": "NO"})
parcels = gpd.read_file("zip://leeds_inspire.zip!Land_Registry_Cadastral_Parcels.gml", columns=["INSPIREID"])
ONS = "https://services1.arcgis.com/ESMARspQHYMw9BZ9/arcgis/rest/services"
wards = gpd.read_file(f"{ONS}/Wards_December_2024_Boundaries_UK_BFC/FeatureServer/0/query"
"?where=LAD24NM%3D%27Leeds%27&outFields=WD24CD&outSR=27700&f=geojson"
).set_crs(27700, allow_override=True)
points = parcels.set_geometry(parcels.representative_point())
ward_of = gpd.sjoin(points, wards, predicate="within")["WD24CD"]
Path("data").mkdir(exist_ok=True)
for code, rows in parcels.join(ward_of, how="inner").groupby("WD24CD"):
rows.drop(columns="WD24CD").to_file(f"data/{code}.gpkg")
files = sorted(Path("data").glob("*.gpkg"))
print(f"{len(files)} ward files, {sum(f.stat().st_size for f in files) / 1e6:.0f} MB")33 ward files, 158 MB
Example 1: the complete incremental job
%%writefile incremental.py
"""Reproject a folder of GeoPackages, processing only what changed."""
import argparse
import hashlib
import inspect
import json
from collections import Counter
from dataclasses import dataclass
from datetime import datetime, timezone
from pathlib import Path
import geopandas as gpd
STATE = Path("state/manifest.json")
CONFIG = {"target_crs": 4326}
def content_hash(path, chunk=1 << 20):
h = hashlib.blake2b(digest_size=16)
with open(path, "rb") as f:
while block := f.read(chunk):
h.update(block)
return h.hexdigest()
def load_state():
return json.loads(STATE.read_text()) if STATE.exists() else {}
def save_state(state):
STATE.parent.mkdir(parents=True, exist_ok=True)
tmp = STATE.with_suffix(".tmp")
tmp.write_text(json.dumps(state, indent=2, sort_keys=True))
tmp.replace(STATE)
def changed_files(src, state, recipe):
for path in sorted(src.rglob("*.gpkg")):
prev = state.get(str(path))
if prev is None:
yield path, "new"
elif prev["recipe"] != recipe:
yield path, "recipe"
elif prev["size"] != path.stat().st_size:
yield path, "size"
elif prev["hash"] != content_hash(path):
yield path, "content"
def recipe_id(config, *functions):
h = hashlib.blake2b(digest_size=8)
h.update(json.dumps(config, sort_keys=True, default=str).encode())
for fn in functions:
h.update(inspect.getsource(fn).encode())
return h.hexdigest()
def guard(work, total, *, threshold=0.5, force=False):
if force or not total or len(work) / total <= threshold:
return work
raise SystemExit(f"{len(work):,}/{total:,} ({len(work) / total:.0%}) look changed — that is unusual. "
f"Check the manifest and the recipe id, then re-run with --force if it really is a rebuild.")
def prune(state, src, out_dir, *, delete_outputs=False):
gone = [k for k in state if not Path(k).exists()]
for key in gone:
entry = state.pop(key)
if delete_outputs and entry.get("output"):
Path(entry["output"]).unlink(missing_ok=True)
return gone
@dataclass
class Result:
ok: bool
rows: int = 0
error: str = ""
def now_iso():
return datetime.now(timezone.utc).isoformat(timespec="seconds")
def output_for(src, src_root, out_root):
return (out_root / src.relative_to(src_root)).with_suffix(".gpkg")
def process(src, src_root, out_root, config):
gdf = gpd.read_file(src)
if gdf.crs is None:
return Result(False, error="no CRS")
gdf = gdf.to_crs(config["target_crs"])
out = output_for(src, src_root, out_root)
out.parent.mkdir(parents=True, exist_ok=True)
tmp = out.with_suffix(".tmp.gpkg")
gdf.to_file(tmp, driver="GPKG")
tmp.replace(out)
return Result(True, rows=len(gdf))
def main(argv=None):
ap = argparse.ArgumentParser()
ap.add_argument("--src", type=Path, default=Path("data"))
ap.add_argument("--out", type=Path, default=Path("output"))
ap.add_argument("--dry-run", action="store_true")
ap.add_argument("--force", action="store_true", help="ignore the manifest")
args = ap.parse_args(argv)
state = {} if args.force else load_state()
recipe = recipe_id(CONFIG, process)
all_files = sorted(args.src.rglob("*.gpkg"))
work = list(changed_files(args.src, state, recipe))
reasons = Counter(r for _, r in work)
print(f"{len(all_files):,} files · {len(work):,} to process {dict(reasons)}")
if args.dry_run:
for path, reason in work[:20]:
print(f" would process {path} ({reason})")
return {"processed": 0, "planned": len(work)}
if state: # a first run legitimately processes everything
work = guard(work, len(all_files), force=args.force)
ok = failed = 0
for path, reason in work:
try:
result = process(path, args.src, args.out, CONFIG)
except Exception as exc: # one bad file must not stop the run
result = Result(False, error=f"{type(exc).__name__}: {str(exc)[:80]}")
if result.ok:
ok += 1
state[str(path)] = {
"hash": content_hash(path), "size": path.stat().st_size,
"mtime": path.stat().st_mtime, "recipe": recipe,
"output": str(output_for(path, args.src, args.out)),
"processed_at": now_iso(), "rows": result.rows,
}
else:
failed += 1
print(f" ✗ {path}: {result.error}")
removed = prune(state, args.src, args.out)
save_state(state)
print(f"done — {ok} processed, {len(all_files)-len(work)} skipped, "
f"{failed} failed, {len(removed)} pruned")
return {"processed": ok, "failed": failed, "skipped": len(all_files) - len(work)}Run it the way a scheduler would, night after night — with one ward's file edited before the third run, and the target CRS changed before the fourth:
import time
import geopandas as gpd
import incremental
def night(label, argv=()):
t0 = time.perf_counter()
try:
counts = incremental.main(["--src", "data", "--out", "output", *argv])
except SystemExit as stop:
print(f" stopped: {stop}")
counts = None
print(f" [{label}: {time.perf_counter() - t0:.1f} s]\n")
return counts
night("first run")
night("second run, nothing changed")
edited = sorted(Path("data").glob("*.gpkg"))[5]
g = gpd.read_file(edited)
g.iloc[:-1].to_file(edited) # the supplier drops one title
night("third run, one ward edited")
incremental.CONFIG["target_crs"] = 3857 # a change to the recipe
night("fourth run, target CRS changed")
night("fifth run, the rebuild confirmed", ["--force"])33 files · 33 to process {'new': 33}
done — 33 processed, 0 skipped, 0 failed, 0 pruned
[first run: 3.8 s]
33 files · 0 to process {}
done — 0 processed, 33 skipped, 0 failed, 0 pruned
[second run, nothing changed: 0.2 s]
33 files · 1 to process {'content': 1}
done — 1 processed, 32 skipped, 0 failed, 0 pruned
[third run, one ward edited: 0.3 s]
33 files · 33 to process {'recipe': 33}
stopped: 33/33 (100%) look changed — that is unusual. Check the manifest and the recipe id, then re-run with --force if it really is a rebuild.
[fourth run, target CRS changed: 0.0 s]
33 files · 33 to process {'new': 33}
done — 33 processed, 0 skipped, 0 failed, 0 pruned
[fifth run, the rebuild confirmed: 4.0 s]
The whole story in five nights. The first run processes all 33 files; the second processes none; the third finds exactly the one file that changed, by content hash; the fourth sees that every file was produced by a different recipe — and the guard refuses, because a run that suddenly wants to redo everything is more often a bug than a decision; the fifth, with --force, does the deliberate rebuild.
The if state: in front of the guard is there because the first version of this job, run for this page, refused its own first run: with an empty manifest every file is "new", 33 of 33 look changed, and the guard does exactly what it was written to do. A guard needs to know what normal looks like, and on the first night there is no normal yet.
--force and --dry-run are both essential in practice, and main takes its arguments as a list so that a test — or the cell above — can call it without touching sys.argv: the first for a deliberate rebuild, the second for answering "what would tonight do?" without doing it.
Example 2: incremental and parallel together
The manifest must be written by one process. Workers return facts; the parent records them.
from concurrent.futures import ProcessPoolExecutor
def run_parallel(work, src_root, out_root, config, workers=8):
updates = {}
with ProcessPoolExecutor(workers) as pool:
futures = {
pool.submit(process_and_fingerprint, p, src_root, out_root, config): p
for p, _ in work
}
for fut in as_completed(futures):
path = futures[fut]
try:
entry = fut.result()
if entry:
updates[str(path)] = entry # collected, not written
except Exception as exc:
print(f" ✗ {path}: {type(exc).__name__}: {exc}")
return updates # parent writes onceTwo processes writing one JSON file interleave and produce a corrupt manifest. Since a corrupt manifest triggers a full rebuild, that bug costs an entire overnight window — worth the discipline of returning data rather than writing from workers.
Example 3: proving it works
%%writefile test_incremental.py
import shutil
from pathlib import Path
import geopandas as gpd
import pytest
import incremental
@pytest.fixture
def three_files(tmp_path, monkeypatch):
"""Three real ward files, in a folder of their own, with a fresh manifest."""
src = tmp_path / "src"
src.mkdir()
for name in sorted(Path("data").glob("*.gpkg"))[:3]:
shutil.copy(name, src / name.name)
monkeypatch.chdir(tmp_path) # manifest and outputs land in tmp_path
return src
def run(src, force=False):
return incremental.main(["--src", str(src), "--out", "out", *(["--force"] if force else [])])
def test_first_run_processes_everything(three_files):
assert run(three_files)["processed"] == 3
def test_second_run_processes_nothing(three_files):
run(three_files)
assert run(three_files)["processed"] == 0
def test_edited_file_is_reprocessed(three_files):
run(three_files)
path = sorted(three_files.glob("*.gpkg"))[1]
gpd.read_file(path).iloc[:-1].to_file(path)
assert run(three_files)["processed"] == 1
def test_recipe_change_is_caught_by_the_guard(three_files, monkeypatch):
run(three_files)
monkeypatch.setitem(incremental.CONFIG, "target_crs", 3857)
with pytest.raises(SystemExit):
run(three_files)
assert run(three_files, force=True)["processed"] == 3
def test_failed_item_is_retried(three_files):
broken = three_files / "broken.gpkg"
broken.write_bytes(b"not a geopackage")
first = run(three_files)
assert first["failed"] == 1
broken.unlink()
shutil.copy(sorted(three_files.glob("*.gpkg"))[0], broken)
assert run(three_files)["processed"] == 1 # retried, not skipped!{sys.executable} -m pytest -q --no-header -p no:cacheprovider --color=no --tb=no -rf test_incremental.py..... [100%]
5 passed in 2.87s
The last two are the ones that catch real regressions, and on this page the last one did. process only returned a failure for the problem its author had thought of — a missing CRS — so a corrupt file raised out of the loop and took the whole run down with it, before the manifest was saved. The try around each item in main is the fix: every failure becomes a recorded failure, retried next time, whatever caused it.
Stale outputs against needless rebuilds
An incremental job is a cache with the outputs as cached values and the manifest as the key store. Everything that makes caches hard applies, and the two failure modes are opposites.
Stale: the key is incomplete, so a change slips through and the job serves old output while reporting success. This is what happens when the check is "does the output exist" and a supplier overwrites a file, or when there is no recipe id and the transformation changes.
Thrashing: the key includes something irrelevant, so everything looks changed and the job does full work while claiming to be incremental. Usually a timestamp or an absolute path leaking into the recipe id.
Stale is far worse, because it is invisible — the counts look normal, no error is raised, and the symptom appears weeks later as a number that stopped moving. Thrashing merely wastes time and shows up in the runtime. That asymmetry is why the recommended default is content hash plus recipe id: it never misses a change, and the guard in step 4 catches the thrashing case before it costs a night.
The ordering rules follow from the same reasoning. Writing state before the work records things that may not happen; writing it non-atomically risks a truncated manifest, which triggers a full rebuild. Both are cheap to get right and expensive to get wrong.
The nicest property of the design is that retry falls out for free. A failed item never gets a manifest entry, so the next run finds it again. There is no retry queue, no failure state to manage, and no way for a failure to be silently forgotten — the absence of evidence is the retry mechanism.
Input hashes, empty files and the manifest lock
- Hash the input, not the output. Some drivers write non-deterministic bytes, so an output hash changes even when the data does not.
- A GeoPackage's bytes can change without the data changing — SQLite reorganises pages. Hashing the file is still the right check for "did the supplier send something different".
- Empty or zero-byte inputs hash consistently and will be marked processed. Validate before recording.
Path.replace()is atomic only on the same filesystem. Put the temp file next to the final one, not in/tmp.- The manifest is not a lock. Two concurrent runs of the same job will both process and both write; use a lock file if that is possible.
- Prune deleted inputs, or the manifest and the output directory drift apart forever.
- Do not commit the manifest. It is run state; it belongs with the outputs and in the backup.
--forceshould still write the manifest, so the next normal run is incremental again.
Internal links
- Incremental processing: how a batch job knows what changed — the concept and the four signals
- The anatomy of a batch job: discover, apply, report — why this lives in discovery
- How to build a resumable batch GIS job in Python — resuming within a run
- How to cache pipeline steps so only changed data is reprocessed — the same idea inside a pipeline
- Idempotency explained: why a GIS job must be safe to re-run — the property that makes skipping safe
- Batch job gives different results each run — when the recipe is less fixed than it looks
- How to add a dry-run mode to a batch GIS script — inspecting the plan
- How to record run metadata and data lineage — the manifest as lineage
FAQ
Is checking the output file's existence enough?
Only if inputs never change after they land. If a supplier can overwrite a file in place, that check silently stops updating and never reports a problem.
How slow is hashing everything?
About 1–2 GB/s from local SSD, much slower over a network share. Use size as a free pre-filter, and mtime as a second one if the volume makes full hashing impractical.
Why does the recipe id matter?
Because the output depends on the code and config, not just the input. Without it, changing the target CRS produces a run that skips every file and reports success.
What if the manifest is deleted?
Everything looks new and the job does a full rebuild — correct but expensive. The guard turns that into a prompt rather than a silent full run.
How do incremental and parallel work together?
Workers return manifest entries; the parent writes the file once. Two processes writing one JSON manifest will corrupt it, and a corrupt manifest costs a full rebuild.
What happens to an item that fails?
It gets no manifest entry, so the next run retries it. Retry is a consequence of the design rather than something you add.
Should I delete outputs whose input disappeared?
For a mirror, yes; for an archive, no. Either way, prune the manifest entry or it grows forever.