How to Process Only the Files That Changed Since the Last Run
Problem statement
The nightly job takes six hours and forty minutes. On a typical night, four files out of nine hundred 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 reprocesses 41 GB to produce 896 outputs that are identical to the ones they replace. The overnight window is nearly full, the downstream sync sees 900 changed files instead of four, and adding a new region will not fit.
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, 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 900 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 loop02:00 discovered 900 files, 4 need processing (3 content, 1 new)
02:03 done — 4 processed, 896 skipped, 0 failed
Six hours and thirty-seven minutes returned to the night.
What the manifest holds
Step-by-step solution
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 9 GB is seconds. Most genuine changes alter the size, so the cheap check resolves the majority without reading anything.
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, hashlib, json
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()
RECIPE = recipe_id(CONFIG, transform, clean_attributes)Without this, changing the target CRS from 27700 to 3857 produces a run that skips all 900 files, 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: 900 items 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.
Code examples
Example 1: the complete incremental job
import argparse, hashlib, json
from collections import Counter
from dataclasses import dataclass
from datetime import datetime, timezone
from pathlib import Path
import geopandas as gpd
@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():
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()
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 0
work = guard(work, len(all_files), force=args.force)
ok = failed = 0
for path, reason in work:
result = process(path, args.src, args.out, CONFIG)
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 1 if failed else 0--force and --dry-run are both essential in practice: 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
def test_first_run_processes_everything(tmp_path, three_files):
n = run(src=three_files, out=tmp_path)
assert n["processed"] == 3
def test_second_run_processes_nothing(tmp_path, three_files):
run(src=three_files, out=tmp_path)
assert run(src=three_files, out=tmp_path)["processed"] == 0
def test_edited_file_is_reprocessed(tmp_path, three_files):
run(src=three_files, out=tmp_path)
append_a_feature(three_files / "b.gpkg")
assert run(src=three_files, out=tmp_path)["processed"] == 1
def test_recipe_change_reprocesses_everything(tmp_path, three_files, monkeypatch):
run(src=three_files, out=tmp_path)
monkeypatch.setitem(CONFIG, "target_crs", 3857)
assert run(src=three_files, out=tmp_path)["processed"] == 3
def test_failed_item_is_retried(tmp_path, three_files, broken_file):
first = run(src=three_files, out=tmp_path)
assert first["failed"] == 1
assert run(src=three_files, out=tmp_path)["processed"] == 1 # retried, not skippedThe last two are the ones that catch real regressions. Skipping after a failure, and failing to invalidate on a code change, are both silent in production and obvious in a test.
Explanation
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.
Edge cases or notes
- 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 six hours.
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.