Using DVC Pipelines for Automated Dataset Snapshots

A declarative dvc.yaml pipeline turns annotation archival from a manual chore into a reproducible, hash-gated stage. DVC normalises all inputs to a consistent coordinate reference system — by default EPSG:4326 — validates spatial integrity, computes SHA-256 checksums for every changed asset, then pushes only the modified chunks to remote storage. The precise geometry, projection, and label schema used in each training run become permanently traceable without zip archives, timestamp-based backups, or guesswork.

Why Annotation Drift Breaks Geospatial Training Pipelines

A single reprojected raster, shifted polygon vertex, or corrected .prj file can silently invalidate months of training metrics without triggering any Git diff on binary assets. Teams that version spatial data only via Git LFS hit two compounding problems: large binary blobs slow every clone, and LFS provides no built-in mechanism to assert geometric consistency before archival. The result is annotation drift — where the dataset a model was trained on differs from the dataset recorded in the experiment tracker, making rollback guesswork. Tracking annotation changes with SHA hashing solves the detection problem; DVC pipelines solve the enforcement problem by making the hash check a mandatory gate before any output artifact is written.

Step-by-Step DVC Pipeline Implementation

DVC snapshot pipeline for geospatial annotations Raw GeoJSON annotation files enter a DVC pipeline stage that validates CRS and geometry, computes SHA-256 checksums, writes a Parquet snapshot manifest, and pushes changed chunks to S3/GCS remote storage. Git receives only lightweight dvc.lock pointer files. DVC pipeline stage: snapshot_annotations Raw GeoJSON annotations/ data/raw/ CRS + Geometry Validation reproject → repair EPSG:4326 SHA-256 Hashing per-file digest + pipeline hash Parquet Snapshot latest_snapshot .parquet S3 / GCS Remote dvc push (changed chunks only) Git repository dvc.lock pointer files only

Step 1 — Install DVC with Your Storage Backend

bash 
# S3 backend
pip install "dvc[s3]>=3.50.0"

# GCS backend
pip install "dvc[gs]>=3.50.0"

# Azure Blob backend
pip install "dvc[azure]>=3.50.0"

# Initialise inside your project root (alongside .git/)
dvc init
git add .dvc .dvcignore
git commit -m "chore: initialise DVC"

Step 2 — Declare the Snapshot Pipeline Stage

Create dvc.yaml at the project root. The deps list is the contract DVC hashes; any change there triggers a re-run:

yaml 
# dvc.yaml
stages:
  snapshot_annotations:
    cmd: >-
      python scripts/validate_and_snapshot.py
        --input  data/raw/annotations/
        --output data/snapshots/
        --target-crs 4326
    deps:
      - data/raw/annotations/
      - scripts/validate_and_snapshot.py
    outs:
      - data/snapshots/latest_snapshot.parquet
      - data/snapshots/metadata.json
    metrics:
      - data/snapshots/pipeline_metrics.json:
          cache: false

Setting cache: false on the metrics file lets DVC track it in Git without pushing it to remote storage on every run.

Step 3 — Write the Validation and Hashing Script

The script below normalises all input annotations to the requested target CRS before hashing, so the stored SHA-256 always reflects a deterministic geometric state rather than the arbitrary projection delivered by individual annotators.

python 
# scripts/validate_and_snapshot.py
"""
Validate CRS + geometry for every GeoJSON file in the input directory,
compute per-file SHA-256 hashes, and write a Parquet snapshot manifest.

Requirements:
    geopandas>=0.14.0
    pyproj>=3.6.0
    shapely>=2.0.0
    pandas>=2.1.0
    pyarrow>=14.0.0
"""
from __future__ import annotations

import argparse
import hashlib
import json
import logging
import pathlib
from datetime import datetime, timezone
from typing import Any

import geopandas as gpd
import pandas as pd

logging.basicConfig(level=logging.INFO, format="%(levelname)s: %(message)s")
log = logging.getLogger(__name__)


def sha256_file(path: pathlib.Path) -> str:
    """Stream-hash a file without loading it fully into memory."""
    digest = hashlib.sha256()
    with path.open("rb") as fh:
        for chunk in iter(lambda: fh.read(65_536), b""):
            digest.update(chunk)
    return digest.hexdigest()


def validate_and_normalise(
    path: pathlib.Path,
    target_epsg: int,
) -> gpd.GeoDataFrame | None:
    """
    Load, validate, and reproject a GeoJSON annotation file.

    Returns None and logs a warning if the file cannot be safely processed.
    """
    try:
        gdf: gpd.GeoDataFrame = gpd.read_file(path)
    except Exception as exc:
        log.error("Cannot read %s: %s", path.name, exc)
        return None

    if gdf.crs is None:
        log.warning("No CRS declared in %s — skipping.", path.name)
        return None

    if gdf.crs.to_epsg() != target_epsg:
        log.info("Reprojecting %s → EPSG:%d", path.name, target_epsg)
        gdf = gdf.to_crs(epsg=target_epsg)

    invalid_mask = ~gdf.geometry.is_valid
    if invalid_mask.any():
        n = int(invalid_mask.sum())
        log.warning("Repairing %d invalid geometries in %s via buffer(0).", n, path.name)
        gdf.loc[invalid_mask, "geometry"] = (
            gdf.loc[invalid_mask, "geometry"].buffer(0)
        )
        still_invalid = ~gdf.geometry.is_valid
        if still_invalid.any():
            log.error(
                "%d geometries in %s could not be repaired — skipping file.",
                int(still_invalid.sum()),
                path.name,
            )
            return None

    return gdf


def main(input_dir: str, output_dir: str, target_crs: int) -> None:
    input_path = pathlib.Path(input_dir)
    output_path = pathlib.Path(output_dir)
    output_path.mkdir(parents=True, exist_ok=True)

    if not input_path.is_dir():
        raise FileNotFoundError(f"Input directory not found: {input_path}")

    records: list[dict[str, Any]] = []

    for geojson in sorted(input_path.glob("*.geojson")):
        gdf = validate_and_normalise(geojson, target_crs)
        if gdf is None:
            continue

        file_hash = sha256_file(geojson)
        records.append(
            {
                "filename": geojson.name,
                "sha256": file_hash,
                "feature_count": len(gdf),
                "crs_epsg": target_crs,
                "geometry_types": sorted(gdf.geometry.geom_type.unique().tolist()),
                "snapshot_utc": datetime.now(timezone.utc).isoformat(),
            }
        )
        log.info("Processed %s  sha256=%s…", geojson.name, file_hash[:12])

    if not records:
        raise ValueError("No valid GeoJSON files were processed — aborting snapshot.")

    df = pd.DataFrame(records)
    parquet_path = output_path / "latest_snapshot.parquet"
    df.to_parquet(parquet_path, index=False)

    pipeline_hash = hashlib.sha256(
        json.dumps(records, sort_keys=True).encode()
    ).hexdigest()
    metadata: dict[str, Any] = {
        "schema_version": "1.1",
        "generated_at": datetime.now(timezone.utc).isoformat(),
        "total_files": len(records),
        "target_crs_epsg": target_crs,
        "pipeline_hash": pipeline_hash,
    }
    (output_path / "metadata.json").write_text(
        json.dumps(metadata, indent=2), encoding="utf-8"
    )

    metrics: dict[str, Any] = {
        "files_processed": len(records),
        "total_features": int(df["feature_count"].sum()),
        "pipeline_hash": pipeline_hash,
    }
    (output_path / "pipeline_metrics.json").write_text(
        json.dumps(metrics, indent=2), encoding="utf-8"
    )

    log.info(
        "Snapshot complete: %d file(s) → %s  pipeline_hash=%s…",
        len(records),
        parquet_path,
        pipeline_hash[:12],
    )


if __name__ == "__main__":
    parser = argparse.ArgumentParser(
        description="Validate and snapshot geospatial annotations."
    )
    parser.add_argument("--input", required=True, help="Raw annotations directory")
    parser.add_argument("--output", required=True, help="Snapshot output directory")
    parser.add_argument(
        "--target-crs",
        type=int,
        default=4326,
        help="EPSG code to normalise all annotations to before hashing (default: 4326)",
    )
    args = parser.parse_args()
    main(args.input, args.output, args.target_crs)

Step 4 — Configure Remote Storage and Run the Pipeline

bash 
# Add a versioned S3 prefix as the default DVC remote
dvc remote add -d spatial-remote s3://your-bucket/dvc-cache
dvc remote modify spatial-remote credentialpath ~/.aws/credentials

# Reproduce the pipeline (no-op if deps are unchanged)
dvc repro

# Push only the new/changed cache chunks to S3
dvc push

# Commit the lightweight .dvc lock files and metrics to Git
git add dvc.lock data/snapshots/pipeline_metrics.json
git commit -m "data: snapshot annotations v$(date +%Y%m%d)"

dvc repro checks every dep hash against what is stored in dvc.lock. If even one .geojson file or the validation script changes, the stage re-runs in full and produces a new set of output hashes. This keeps the repository under 1 MB while tracking terabytes of spatial data in the remote cache.

For raster assets such as multi-gigabyte GeoTIFF mosaics, avoid local cache bloat by using external tracking:

bash 
dvc add --external s3://your-bucket/raw-imagery/mosaic_2024.tif

DVC writes a .dvc pointer file containing only the hash — the binary never enters the local cache.

Step 5 — Automate Snapshots in CI/CD

A GitHub Actions workflow that gates pull requests on pipeline success prevents annotation drift from reaching production training jobs. The metadata.json manifest output integrates with experiment trackers such as MLflow — see Preserving Metadata Across Dataset Versions for how to embed CRS, geometry type, and label schema into versioned manifests consumed by training scripts.

yaml 
# .github/workflows/annotation-snapshot.yml
name: Annotation Snapshot

on:
  pull_request:
    paths:
      - "data/raw/annotations/**"
      - "scripts/validate_and_snapshot.py"
      - "dvc.yaml"

jobs:
  snapshot:
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v4

      - uses: actions/setup-python@v5
        with:
          python-version: "3.11"

      - name: Install dependencies
        run: pip install "dvc[s3]>=3.50.0" geopandas>=0.14.0 pyarrow>=14.0.0

      - name: Configure DVC remote
        env:
          AWS_ACCESS_KEY_ID: $
          AWS_SECRET_ACCESS_KEY: $
        run: |
          dvc remote add -d spatial-remote s3://your-bucket/dvc-cache
          dvc pull --run-cache

      - name: Reproduce snapshot pipeline
        run: dvc repro --pull

      - name: Push updated cache artifacts
        if: github.event_name == 'push'
        run: dvc push

Spatial Parameters and Pipeline Thresholds

Parameter Value Purpose
--target-crs 4326 (default) EPSG code for normalisation before hashing
DVC remote type s3, gs, azure, ssh Backend for content-addressable cache
Chunk size (hash) 65 536 bytes Balances memory use and I/O throughput for large .geojson files
Parquet compression snappy (pandas default) Efficient columnar storage for the manifest
cache: false (metrics) — Keeps pipeline_metrics.json in Git rather than S3
--external flag on .tif paths Avoids copying multi-GB rasters into local DVC cache

Common Errors and Fixes

ERROR: No valid GeoJSON files processed. : Every file in data/raw/annotations/ has a missing CRS or unrepairable geometries. Run ogrinfo -al -so <file>.geojson to confirm CRS is declared, and audit geometry health with python -c "import geopandas as gpd; gdf=gpd.read_file('<file>'); print(gdf.geometry.is_valid.value_counts())" before the pipeline run.

dvc repro reports stage is cached after editing a .geojson file : DVC hashes file content, not modification time. If content did change, confirm the file is listed under deps in dvc.yaml — DVC tracks directories by recursively hashing their contents, so parent-directory entries usually resolve this. Run dvc status --verbose to compare stored vs current hashes.

pyproj.exceptions.CRSError: Invalid projection : The annotation tool exported a .geojson with a non-standard or missing crs member. Use pyproj.CRS.from_user_input(gdf.crs) for a fuzzy match, or force gdf.crs = pyproj.CRS.from_epsg(4326) when the source CRS is known from project documentation.

dvc push transfers unchanged chunks repeatedly : The remote is misconfigured with a path prefix that varies between runs (e.g. a date-stamped folder). Pin a stable url in .dvc/config and use dvc gc --cloud -w sparingly to prune truly unreferenced objects rather than deleting valid cache entries.

This workflow is one component of the broader Tracking Annotation Changes with SHA Hashing strategy for deterministic spatial data lineage.

Related

Related in Tracking Annotation Changes with SHA Hashing

Computing Stable Content Hashes for COGs
A raw file SHA of a Cloud-Optimized GeoTIFF changes on almost every re-export, even when not a single pixel moved. GeoTIFF headers carry volatile bytes — a creation timestamp, a GDAL version tag, TIFF tag ordering, and shifting IFD offsets — so re-running the identical export on a newer library or a different machine r