Writing Hive-Partitioned Parquet for Sensor Archives

An archive is only useful if it can still answer questions. Dumping partitions to compressed CSV meets the letter of a retention policy and fails its purpose: three years later someone needs the readings from one site for one month, and the only way to get them is to decompress and scan everything. Parquet with sensible partitioning gives you the same storage saving and the ability to read one month of one site without touching the rest — which is what makes an archive a tier rather than a tomb. This guide writes that archive, and the verification step that must pass before the source data is dropped, for time-series storage partitioning and retention.

Partition Layout Is a Query Decision

Hive partitioning encodes column values in the directory path — year=2026/month=08/ — so a reader filtering on those columns skips whole directories without opening a file. The decision is which columns to encode, and it follows entirely from how the archive will be read.

Environmental archives are read in two ways. “Give me everything for this period” — a reanalysis, a restatement, an open-data publication. And “give me this site’s history” — a site-specific investigation. The first is served by time partitioning. The second is served by row-group statistics rather than by partitioning, because Parquet stores per-row-group min and max for every column, and a reader with a device_id filter skips row groups whose range excludes it.

That asymmetry is why partitioning by device is a mistake. It optimises the query that row-group pruning already handles, at the cost of exploding the file count: 200 devices times 36 months is 7 200 directories, and inside each one a file too small to compress well.

Partition layout decides file count, not query speed Matrix of four Hive partition layouts against the number of directories produced for three years of a 200-sensor network, the typical file size, and what actually serves a per-device query. Partition layout decides file count, not query speed Directories File size Device query served by year 3 3.5 GB row-group statistics year / month 36 290 MB row-group statistics year / month / day 1 095 9 MB row-group statistics year / month / device 7 200 1.5 MB the directory — badly
Only the second row balances both. Partitioning by device optimises the query that statistics already handle, at the cost of thousands of files too small to compress.

Production-Ready Implementation 

# python 3.11 · pyarrow==15.0.2 · pandas==2.2.2 · psycopg[binary]==3.1.18
from __future__ import annotations

import hashlib
import pathlib

import pandas as pd
import pyarrow as pa
import pyarrow.dataset as ds
import pyarrow.parquet as pq

SCHEMA_VERSION = "2"

ARCHIVE_SCHEMA = pa.schema(
    [
        ("device_id", pa.string()),
        ("observed_at", pa.timestamp("us", tz="UTC")),
        ("metric", pa.string()),
        ("value", pa.float64()),
        ("qc_flag", pa.int16()),
        ("uncertainty", pa.float32()),
        ("registry_version", pa.int32()),
    ],
    metadata={b"schema_version": SCHEMA_VERSION.encode()},
)


def write_month(frame: pd.DataFrame, root: pathlib.Path) -> pathlib.Path:
    """Write one month of readings as Hive-partitioned Parquet.

    Sorting by (device_id, observed_at) before writing is what makes row-group
    statistics useful: an unsorted file has every device in every row group, so
    a device filter prunes nothing at all.
    """
    frame = frame.sort_values(["device_id", "observed_at"]).copy()
    frame["year"] = frame["observed_at"].dt.year.astype("int32")
    frame["month"] = frame["observed_at"].dt.month.astype("int32")

    table = pa.Table.from_pandas(frame, schema=_with_partitions(ARCHIVE_SCHEMA),
                                 preserve_index=False)
    pq.write_to_dataset(
        table,
        root_path=str(root),
        partition_cols=["year", "month"],
        compression="zstd",
        compression_level=3,
        row_group_size=1_000_000,       # ~1M rows: big enough to compress, small enough to skip
        use_dictionary=["device_id", "metric"],
        write_statistics=True,
        existing_data_behavior="overwrite_or_ignore",
    )
    return root


def _with_partitions(schema: pa.Schema) -> pa.Schema:
    return schema.append(pa.field("year", pa.int32())).append(pa.field("month", pa.int32()))

Three settings do most of the work. Dictionary encoding on device_id and metric collapses repeated strings to small integers, which on sensor data is the single largest compression win. row_group_size of a million rows balances compression against skip granularity. And the sort before writing is what makes the statistics meaningful — without it, every row group spans every device and no pruning is possible.

Verification before deletion is not optional:

def verify_archive(root: pathlib.Path, source_frame: pd.DataFrame) -> dict:
    """Read the archive back and compare against the source. Raises on any mismatch.

    Compares row count and a stable checksum over the value column, so a
    truncated write or a silently dropped partition cannot pass.
    """
    dataset = ds.dataset(str(root), format="parquet", partitioning="hive")
    archived = dataset.to_table(
        columns=["device_id", "observed_at", "metric", "value"]
    ).to_pandas()

    if len(archived) != len(source_frame):
        raise RuntimeError(f"row count mismatch: {len(archived)} archived, {len(source_frame)} source")

    def checksum(df: pd.DataFrame) -> str:
        ordered = df.sort_values(["device_id", "observed_at", "metric"])
        payload = ordered["value"].round(6).to_numpy().tobytes()
        return hashlib.sha256(payload).hexdigest()

    a, b = checksum(archived), checksum(source_frame)
    if a != b:
        raise RuntimeError(f"value checksum mismatch: {a[:12]} vs {b[:12]}")
    return {"rows": len(archived), "checksum": a, "schema_version": SCHEMA_VERSION}

Reading it back later is a filter, not a scan:

def read_site_month(root: pathlib.Path, device_id: str, year: int, month: int) -> pd.DataFrame:
    """One device, one month — partition pruning plus row-group statistics."""
    dataset = ds.dataset(str(root), format="parquet", partitioning="hive")
    return dataset.to_table(
        filter=(ds.field("year") == year)
        & (ds.field("month") == month)
        & (ds.field("device_id") == device_id)
    ).to_pandas()
Compression codec against size and read speed Grouped bar chart comparing uncompressed, snappy and zstd Parquet on file size for one month of readings and on the time to read one device back. Compression codec against size and read speed 0 500 1000 uncompressed snappy zstd level 3 zstd level 9 MB and milliseconds size (MB) read one device (ms)
Level 3 is the landing spot: nearly all the compression of level 9 without the decompression cost, on data written once and read rarely.

Parameter Tuning Guide

Setting Value Why
partition_cols ["year", "month"] matches the dominant access pattern; bounds file count
compression zstd, level 3 8–15× on sensor data; fast enough to stay IO-bound
row_group_size 1 000 000 rows ~40–80 MB per group after compression
use_dictionary device_id, metric repeated low-cardinality strings compress enormously
Sort order before write (device_id, observed_at) makes row-group statistics prunable
Target file size 128–512 MB below 64 MB, listing overhead dominates on object storage
Network size Rows/month Parquet size/month Files/month
20 × 2, 5 min 350 k 3 MB 1
200 × 4, 1 min 34.5 M 290 MB 1–2
2 000 × 4, 1 min 345 M 2.9 GB 8–12
20 000 × 6, 1 min 5.2 B 44 GB 100–150

Compare the second row against its row-store footprint — roughly 3.3 GB per month in PostgreSQL including indexes — for the 11× reduction that justifies the tier.

Write, verify, then — and only then — drop Flow diagram of the archive path: sort, write partitioned Parquet, read it back and compare row count and checksum against the source, record the result, and only then allow the retention job to drop the partition. Write, verify, then — and only then — drop Sort by device, time makes statistics prunable before writing Write partitioned year / month zstd, 1M row groups Read back + checksum rows and values raises on mismatch Record verification archive_log entry unblocks retention A verification whose result is not stored has not really happened — the retention job reads that record, not a log line.
Sorting is what makes the archive queryable; verification is what makes deleting the source safe. Skipping either produces a compressed tomb.

Verification and Testing 

# python 3.11 · pyarrow==15.0.2 · pytest==8.2.0
def test_round_trip_preserves_every_row_and_value(tmp_path, month_frame):
    write_month(month_frame, tmp_path / "reading")
    result = verify_archive(tmp_path / "reading", month_frame)
    assert result["rows"] == len(month_frame)


def test_a_truncated_archive_fails_verification(tmp_path, month_frame):
    write_month(month_frame.iloc[:-100], tmp_path / "reading")
    with pytest.raises(RuntimeError, match="row count mismatch"):
        verify_archive(tmp_path / "reading", month_frame)


def test_device_filter_reads_a_fraction_of_the_bytes(tmp_path, month_frame):
    """Sorting plus statistics must actually prune — otherwise the layout is pointless."""
    write_month(month_frame, tmp_path / "reading")
    dataset = ds.dataset(str(tmp_path / "reading"), format="parquet", partitioning="hive")

    everything = dataset.to_table(columns=["value"]).num_rows
    one_device = dataset.to_table(
        columns=["value"], filter=ds.field("device_id") == month_frame["device_id"].iloc[0]
    ).num_rows
    assert one_device < everything / 10


def test_older_schema_version_still_reads(tmp_path, v1_frame):
    """A file written before a column was added must still load, with nulls."""
    write_month(v1_frame, tmp_path / "reading")
    table = ds.dataset(str(tmp_path / "reading"), format="parquet", partitioning="hive").to_table()
    assert "uncertainty" in table.column_names

Gotchas

Writing without sorting. Row-group statistics become useless, every query reads every group, and the archive performs like compressed CSV with extra steps.

Partitioning by day or by device. The small-file problem is severe on object storage, where listing a prefix with ten thousand keys costs real time and money. Year and month is almost always right.

Timestamps written as strings. They compress poorly, sort lexically rather than temporally, and lose the timezone. Use timestamp('us', tz='UTC').

Deleting the source before verifying the archive. The whole reason verify_archive returns a checksum rather than a boolean is so the retention log can record it. A verification whose result is not stored has not really happened.

Assuming the archive is immutable. A calibration correction may require restating archived months. Keep the write path idempotent — overwrite_or_ignore on a whole partition — so a restated month replaces cleanly rather than appending duplicates.

FAQ

How should the archive be partitioned?

By year and month, and nothing finer for most networks. Partitioning by day multiplies file count by thirty for no query benefit once row groups are sized properly, and partitioning by device produces the small-file problem in its worst form — thousands of files a few kilobytes each, where listing costs more than reading.

Which compression codec?

zstd at level 3 for archives. It compresses sensor data 8 to 15 times, decompresses fast enough that queries are IO-bound rather than CPU-bound, and is supported by every current Parquet reader. Snappy is faster to write and roughly 30% larger, which is the wrong trade for data written once and read rarely.

What happens when the schema changes?

Parquet tolerates added columns: older files simply lack them and readers return null. Removing or retyping a column is what breaks, so treat the archive schema as append-only and record a schema version in the file metadata so a reader can tell which generation it is looking at.