Spatial Epi Public Health GIS

Partitioning a GeoParquet Surveillance Archive

This guide, part of Cloud-Native Geospatial Storage for Surveillance, turns the partitioning decision from a matter of taste into a short measurement exercise with a defensible answer.

Problem Context & Constraints

Partitioning trades two costs against each other. Finer partitions mean a selective query touches fewer bytes; they also mean more files, and every file costs a listing entry, a metadata read and a request. On object storage the request cost dominates surprisingly early: a query that must open twelve thousand small files is slower than one that scans a few large ones, even though it reads less data.

The trade is quantifiable. Given a candidate scheme you can compute the file count it produces over the archive’s lifetime, and you can measure the bytes a representative query reads under it. Two numbers, two query shapes, and the answer falls out.

Three constraints bound the search. The partition key must appear in the predicate — partitioning by a column nobody filters on costs files and buys nothing. The pipeline’s write pattern sets a floor, because a nightly run writing one file per partition per night creates at least that many files. And retention must be expressible, so any column the retention policy refers to should be a partition key.

The Two Costs Move in Opposite Directions Two curves plotted against partition grain from coarse on the left to fine on the right. Bytes read per selective query falls steeply as the grain becomes finer, then flattens once the partition is already smaller than a single query's scope. File count rises, slowly at first and then very fast, because each additional key multiplies the partitions. The total cost curve, combining request overhead with transfer, has a minimum in the middle, and the chart marks it as the grain to choose. The best grain is a minimum, not an extreme bytes read file count total cost choose here yearly monthly by county daily by tract partition grain Measure both curves on your own archive; the minimum moves with jurisdiction count

Prerequisites

  • python 3.11
  • geopandas 0.14.4, pyarrow 16.1.0, duckdb 1.0.0, pandas 2.2.2
  • A sample of the archive large enough to be representative, or realistic row counts per date and jurisdiction
  • The two or three query shapes the system actually serves, written as SQL
  • The pipeline’s write cadence, which sets the minimum file count

Step-by-Step Solution

Enumerate candidate schemes, compute their file counts analytically, measure bytes read for each query shape on a sample, and pick the minimum of the combined cost.

# Evaluate candidate partition schemes by file count and measured bytes read.
# Pinned: geopandas==0.14.4, pyarrow==16.1.0, duckdb==1.0.0, pandas==2.2.2
import itertools
import logging
import os

import pandas as pd

logging.basicConfig(level=logging.INFO, format="%(asctime)s %(levelname)s %(message)s")
log = logging.getLogger("storage.partition")

REQUEST_COST_BYTES = 256_000     # empirical cost-equivalent of one object request


def projected_file_count(cardinalities: dict, scheme: tuple,
                         writes_per_partition: int = 1) -> int:
    """How many files a scheme produces over the archive's lifetime."""
    n = writes_per_partition
    for col in scheme:
        n *= cardinalities[col]
    return int(n)


def measure_bytes_read(root: str, sql: str) -> dict:
    """Bytes actually read for a query, using the engine's own profiling."""
    import duckdb
    con = duckdb.connect()
    con.execute("INSTALL spatial; LOAD spatial;")
    con.execute("PRAGMA enable_profiling='json'; PRAGMA profiling_output='/dev/null';")
    view = f"read_parquet('{os.path.join(root, '**', '*.parquet')}', hive_partitioning=true)"
    con.execute(sql.replace("{archive}", view)).fetchall()
    # Fall back to on-disk accounting where the engine does not report bytes.
    files = [os.path.join(dp, f) for dp, _, fs in os.walk(root)
             for f in fs if f.endswith(".parquet")]
    return {"files_present": len(files),
            "archive_bytes": sum(os.path.getsize(f) for f in files)}


def evaluate_schemes(cardinalities: dict, candidate_cols: list[str],
                     queries: dict, sample_roots: dict,
                     writes_per_partition: int = 1) -> pd.DataFrame:
    """Score every candidate scheme on file count and per-query bytes."""
    rows = []
    for r in range(1, len(candidate_cols) + 1):
        for scheme in itertools.permutations(candidate_cols, r):
            files = projected_file_count(cardinalities, scheme, writes_per_partition)
            root = sample_roots.get(scheme)
            if root is None:
                rows.append({"scheme": "/".join(scheme), "files": files,
                             "bytes_total": None, "cost": None})
                continue
            total_bytes = 0
            for qname, sql in queries.items():
                m = measure_bytes_read(root, sql)
                total_bytes += m["archive_bytes"]
            # Request overhead is what makes a fine scheme expensive.
            cost = total_bytes + files * REQUEST_COST_BYTES
            rows.append({"scheme": "/".join(scheme), "files": files,
                         "bytes_total": total_bytes, "cost": cost})
    df = pd.DataFrame(rows).sort_values("cost", na_position="last")
    log.info("evaluated %d schemes; best=%s files=%d",
             len(df), df.iloc[0]["scheme"], int(df.iloc[0]["files"]))
    return df


def compact_partition(partition_dir: str, target_bytes: int = 512 * 1024 * 1024) -> int:
    """Merge the many small files a nightly writer leaves into a few large ones."""
    import pyarrow.parquet as pq
    import pyarrow as pa
    files = sorted(f for f in os.listdir(partition_dir) if f.endswith(".parquet"))
    if len(files) <= 1:
        return 0
    tables = [pq.read_table(os.path.join(partition_dir, f)) for f in files]
    merged = pa.concat_tables(tables, promote_options="permissive")
    out = os.path.join(partition_dir, "compacted-000.parquet.tmp")
    pq.write_table(merged, out, compression="zstd", row_group_size=100_000)
    os.replace(out, out[:-4])
    for f in files:                       # originals removed only after the swap
        os.remove(os.path.join(partition_dir, f))
    log.info("compacted %s: %d files -> 1 (%d rows)", partition_dir, len(files),
             merged.num_rows)
    return len(files)

The request-cost term is what makes the evaluation realistic. Comparing schemes on bytes read alone always prefers the finest grain, which is why intuition tends toward over-partitioning. Adding a cost-equivalent per file — measurable on your own storage by timing a query that opens many small objects — produces a minimum that matches what the system actually experiences.

Compaction is the other half of the answer, because it decouples the write pattern from the read pattern. A nightly writer can produce one file per partition per night, and a weekly compaction can merge them into one, so the archive has a fine write grain and a coarse read grain without either compromising.

Write Fine, Read Coarse One partition shown over a week. Each night the pipeline adds a small file, so by the end of the week the partition holds seven small objects and a query opening it makes seven requests. A weekly compaction reads all seven and writes one large file, removing the originals only after the replacement is in place. The partition then holds a single object with tight row-group statistics, and the following week the cycle repeats. The note records that provenance is preserved because the run identifier travels as a column rather than as a filename. Compaction decouples the write cadence from the read cost during the week 7 objects, 7 requests per query weekly compaction after compaction one object, tight statistics 1 request, same rows Provenance survives because the run identifier is a column, not a filename

Validation & Edge Cases

A partition key nobody filters on. Check the actual query log rather than assuming. A key that appears in no predicate multiplies file count for nothing and should be an ordinary column.

Highly skewed partitions. One large county produces a partition orders of magnitude bigger than the rest, so the scheme is balanced on paper and not in practice. Consider a secondary split for the skewed key, or accept it and size row groups accordingly.

A scheme chosen before the system has a query log. A new archive has no evidence about which predicates will be used, and the temptation is to partition defensively on everything. Start with the coarsest scheme that satisfies retention and access control, keep the other candidate keys as ordinary sorted columns, and revisit once real queries exist. Repartitioning later is a bounded rewrite; unpicking a millions-of-files layout is not.

Partition keys with changing cardinality. A jurisdiction key looks stable until a state reorganises its reporting regions, and a monthly key grows forever. Project the file count over the archive’s expected lifetime rather than its current size, because a scheme that is comfortable at two years of data can be unusable at ten.

Measurements taken on a warm cache. Repeating the same query while tuning reads from the operating system’s page cache and reports a time that production will never see. Clear the cache between measurements, or measure against remote storage directly, which is where the archive will actually live.

Compaction losing provenance. Merging files must preserve the run identifier column, not the file names. If provenance lived in the filename, compaction destroys it — which is why it belongs in a column.

Compaction racing the writer. A compaction that runs while the nightly job writes can miss a file or delete one mid-read. Schedule compaction outside the write window, and always write the merged file before removing the originals.

Schema differences between files in one partition. Files written before a column was added merge with nulls under a permissive promotion, which is usually right and occasionally masks a genuine mismatch. Log the schema union and review changes.

INFO storage.partition evaluated 15 schemes; best=event_month/jurisdiction files=4032
INFO storage.partition compacted /archive/event_month=2026-08/jurisdiction=06075: 31 files -> 1 (418220 rows)

Skew is the reason a scheme that is balanced on paper can still perform badly, and a single large jurisdiction shows it clearly:

One County Can Outweigh All the Others Monthly partition sizes for twelve jurisdictions. Eleven rural and suburban counties produce partitions of broadly similar, modest size. One metropolitan county produces a partition many times larger than any other, so a query touching it reads far more than the average and row-group sizing tuned for the typical partition is wrong for it. A secondary split on a finer key for that county alone, or row groups sized per partition, restores balance. Balanced on paper, skewed in practice metropolitan county split further, or size row groups for it one monthly partition per jurisdiction measure partition sizes before trusting a scheme's arithmetic

Compliance Notes

Record the chosen scheme with the measurements that justified it, the projected file count, the compaction schedule and the retention policy expressed against the partition keys. Where a jurisdiction key exists for access-control reasons rather than performance ones, record that too, because it will otherwise look like an over-partitioning mistake to whoever tunes the archive next.

Compaction has one compliance implication worth stating: it rewrites files, so any file-level retention or legal-hold marker attached to an original object does not survive it. Where a partition is under hold, exclude it from compaction and record the exclusion, rather than discovering after the fact that the held objects were replaced by a merged one with a new creation date.