Append-Only Audit Storage with Parquet on Object Storage

A drinking-water compliance record has to survive audits, litigation, and staff turnover long after the shift that produced it, and the cheapest durable home for that history is columnar Parquet written once to object storage and then frozen. This page is the object-storage reference implementation inside the parent Audit Trail & Data Lineage Storage Patterns section, aimed at the Python engineers who run the compliance pipeline and the environmental staff who must produce, on demand, an unaltered record of every reading and every violation decision. The design goal is narrow and strict: each batch of audit events is written as date-partitioned Parquet into an S3 or Cloudflare R2 bucket, protected by Object Lock so the bytes physically cannot be overwritten or deleted for the retention period, and accompanied by a per-batch manifest that records the row count, byte size, and a content hash so any later reader can prove the partition is intact. That immutability guarantee is what makes the store trustworthy where a mutable table cannot, and it complements the row-level guarantees described in Building Append-Only Audit Logs in PostgreSQL.

Prerequisites & Environment Setup

The implementation targets Python 3.10+, pyarrow for columnar writing, and boto3 as the S3-compatible client. Cloudflare R2 speaks the S3 API, so the same boto3 code drives either backend; you only change the endpoint URL and credentials. The one hard requirement that is not a Python package is a bucket created with Object Lock enabled at creation time — you cannot retrofit Object Lock onto an existing bucket, and a bucket without it silently ignores every retention header you send. Create the bucket first with versioning and object lock turned on, then point the code below at it.

Credentials and the endpoint belong in the environment, never in source, and the writer principal should hold s3:PutObject and s3:PutObjectRetention but deliberately not s3:BypassGovernanceRetention; separating those rights is part of the least-privilege posture covered in Security Boundary Design.

python3 -m venv .venv && source .venv/bin/activate
pip install "pyarrow==16.1.*" "boto3==1.34.*" "pydantic==2.7.*"
# Endpoint + creds via environment (R2 shown; omit endpoint_url for AWS S3)
export AWS_S3_ENDPOINT="https://<accountid>.r2.cloudflarestorage.com"
export AWS_ACCESS_KEY_ID="..." AWS_SECRET_ACCESS_KEY="..."
export AUDIT_BUCKET="wq-compliance-audit"

Step-by-Step Implementation

The pipeline has three moving parts: a writer that serializes a batch to a partitioned Parquet dataset, an uploader that pushes each part under Object Lock and emits a manifest, and a reader that reconstructs a compliance window by listing partitions. The data flow below is worth fixing in your head before reading the code, because every later gotcha traces back to one of these hops.

Step 1 — Serialize a batch to a date-partitioned Parquet dataset

Audit events arrive as a list of dictionaries — one per compliance-relevant action, such as a reading accepted into the historian or a violation decision emitted by the rule engine. Convert them to a pyarrow.Table with an explicit schema (never infer types on audit data, because a single null column silently changes the Parquet schema between batches) and derive a partition column from the event timestamp. Partitioning by year/month/day keeps each leaf directory small enough to list quickly while mapping cleanly onto the calendar windows that reporting rules use.

import hashlib
import io
from datetime import date, datetime, timezone
from typing import Any

import pyarrow as pa
import pyarrow.parquet as pq

AUDIT_SCHEMA = pa.schema([
    ("event_id", pa.string()),
    ("event_ts", pa.timestamp("us", tz="UTC")),
    ("actor", pa.string()),
    ("action", pa.string()),
    ("subject_id", pa.string()),
    ("payload", pa.string()),      # canonical JSON of the audited record
    ("prev_hash", pa.string()),    # chain link to the previous event
])


def build_partition(events: list[dict[str, Any]]) -> tuple[pa.Table, date]:
    """Validate a same-day batch into a typed Arrow table."""
    if not events:
        raise ValueError("refusing to write an empty audit batch")
    days = {e["event_ts"].astimezone(timezone.utc).date() for e in events}
    if len(days) != 1:
        raise ValueError(f"batch spans multiple UTC days: {sorted(days)}")
    table = pa.Table.from_pylist(events, schema=AUDIT_SCHEMA)
    return table, days.pop()


def serialize_parquet(table: pa.Table) -> bytes:
    """Write one Parquet part into memory with deterministic settings."""
    buf = io.BytesIO()
    pq.write_table(
        table, buf,
        compression="zstd",
        use_dictionary=True,
        write_statistics=True,
        version="2.6",
    )
    return buf.getvalue()

Writing the part into an in-memory buffer rather than straight to a filesystem path lets the next step compute a content hash over the exact bytes that will be stored, with no temporary file to leak or race on.

Step 2 — Upload under Object Lock and emit a per-batch manifest

Each part is uploaded once, to a deterministic key built from its partition date and content hash, with an Object Lock retention header. In COMPLIANCE mode no principal — not even the account root — can delete or overwrite the object until the retain-until instant passes. That instant is fixed at write time from the retention period $R$:

where $R$ is expressed in whole years to match the record-retention floors in 40 CFR Part 141. The manifest is a small JSON object written to a sibling key; it records the row count, byte size, and the SHA-256 content hash $h = \text{SHA256}(b)$ over the part bytes $b$, so integrity can be re-checked later without re-reading the whole dataset. That per-part digest is the anchor for the deeper checks in Verifying Record Integrity with Cryptographic Checksums.

import json
import os
from datetime import timedelta

import boto3

s3 = boto3.client("s3", endpoint_url=os.environ.get("AWS_S3_ENDPOINT"))
BUCKET = os.environ["AUDIT_BUCKET"]


def partition_prefix(part_day: date) -> str:
    return f"audit/year={part_day:%Y}/month={part_day:%m}/day={part_day:%d}"


def commit_partition(part_bytes: bytes, table: pa.Table, part_day: date,
                     retention_years: int = 5) -> dict[str, Any]:
    content_hash = hashlib.sha256(part_bytes).hexdigest()
    committed = datetime.now(timezone.utc)
    retain_until = committed + timedelta(days=365 * retention_years)
    prefix = partition_prefix(part_day)
    data_key = f"{prefix}/part-{content_hash[:16]}.parquet"

    s3.put_object(
        Bucket=BUCKET, Key=data_key, Body=part_bytes,
        ContentType="application/vnd.apache.parquet",
        ObjectLockMode="COMPLIANCE",
        ObjectLockRetainUntilDate=retain_until,
        ChecksumSHA256=hashlib.sha256(part_bytes).digest().hex(),
    )

    manifest = {
        "data_key": data_key,
        "partition": prefix,
        "row_count": table.num_rows,
        "byte_size": len(part_bytes),
        "content_sha256": content_hash,
        "committed_at": committed.isoformat(),
        "retain_until": retain_until.isoformat(),
        "retention_mode": "COMPLIANCE",
    }
    s3.put_object(
        Bucket=BUCKET, Key=f"{data_key}.manifest.json",
        Body=json.dumps(manifest, sort_keys=True).encode(),
        ContentType="application/json",
        ObjectLockMode="COMPLIANCE",
        ObjectLockRetainUntilDate=retain_until,
    )
    return manifest

Keying the object by its own content hash makes the write idempotent: a retried upload of the same batch lands on the same key and, because the bytes are identical, does not violate the lock. Locking the manifest with the same retain-until date as the part it describes prevents the subtle attack of altering data while quietly rewriting the record that vouches for it.

Step 3 — Read a compliance window by listing partitions

Producing a report or answering an auditor means selecting every part whose partition date falls inside a window, then reading only those. Because the prefix encodes the date, you list a bounded set of keys instead of scanning the whole bucket, and you verify each part against its manifest before trusting a single row.

def iter_window(start: date, end: date):
    """Yield (data_key, manifest) for every committed part in [start, end]."""
    paginator = s3.get_paginator("list_objects_v2")
    day = start
    while day <= end:
        prefix = partition_prefix(day)
        for page in paginator.paginate(Bucket=BUCKET, Prefix=prefix):
            for obj in page.get("Contents", []):
                key = obj["Key"]
                if not key.endswith(".parquet"):
                    continue
                mf = s3.get_object(Bucket=BUCKET, Key=f"{key}.manifest.json")
                yield key, json.loads(mf["Body"].read())
        day += timedelta(days=1)


def read_window(start: date, end: date) -> pa.Table:
    tables = []
    for data_key, manifest in iter_window(start, end):
        body = s3.get_object(Bucket=BUCKET, Key=data_key)["Body"].read()
        if hashlib.sha256(body).hexdigest() != manifest["content_sha256"]:
            raise IOError(f"content hash mismatch for {data_key}")
        table = pq.read_table(io.BytesIO(body))
        if table.num_rows != manifest["row_count"]:
            raise IOError(f"row count mismatch for {data_key}")
        tables.append(table)
    return pa.concat_tables(tables) if tables else AUDIT_SCHEMA.empty_table()

The reader is where the manifest earns its keep: a byte read that does not match the recorded hash, or a row count that has drifted, raises immediately rather than feeding a corrupted record into a report. Downstream, that verified table is the trustworthy source for the reconciliation queries the parent Audit Trail & Data Lineage Storage Patterns section describes.

Configuration Reference

The parameters below are the knobs you actually tune per deployment. Treat the partition keys and retention mode as versioned configuration, decided once with the compliance team and then held constant, because changing them mid-stream fractures the layout that the reader depends on.

Retention modes at a glance

Verification & Testing

Prove two properties before trusting the store: a round trip preserves every row and the recorded hash, and a tampered byte is caught. The test below exercises the pure serialize-and-hash path with no network, so it runs in CI, and asserts that a single flipped byte breaks the digest the reader checks.

from datetime import datetime, timezone


def _sample_events():
    ts = datetime(2026, 7, 17, 9, 30, tzinfo=timezone.utc)
    return [{
        "event_id": "e-001", "event_ts": ts, "actor": "ingest",
        "action": "reading.accepted", "subject_id": "TURB-07",
        "payload": '{"ntu": 0.12}', "prev_hash": "0" * 64,
    }]


def test_roundtrip_preserves_rows_and_hash():
    table, part_day = build_partition(_sample_events())
    part = serialize_parquet(table)
    digest = hashlib.sha256(part).hexdigest()

    read_back = pq.read_table(io.BytesIO(part))
    assert read_back.num_rows == table.num_rows == 1
    assert hashlib.sha256(part).hexdigest() == digest  # deterministic bytes


def test_tampered_byte_breaks_digest():
    table, _ = build_partition(_sample_events())
    part = serialize_parquet(table)
    tampered = bytearray(part)
    tampered[-1] ^= 0x01
    assert hashlib.sha256(bytes(tampered)).hexdigest() != hashlib.sha256(part).hexdigest()

Acceptance criteria before promoting the store to production:

• The bucket was created with Object Lock enabled; a retention header on a test object is honored, not ignored.
• Every committed part has a sibling .manifest.json with matching row_count, byte_size, and content_sha256.
• The reader raises on a hash or row-count mismatch instead of returning partial data.
• The writer principal lacks s3:BypassGovernanceRetention; a delete attempt on a locked object is denied.
• retain_until on every object is at least the regulatory retention floor from the commit date.
• A same-batch re-upload lands on the identical content-addressed key and does not error against the lock.
• Empty batches are rejected at build_partition, never written as zero-row parts.

Troubleshooting & Gotchas

• Retention headers appear to do nothing. Object Lock cannot be enabled after the fact. If ObjectLockRetainUntilDate is silently accepted but deletes still succeed, the bucket was created without lock support — recreate it with lock enabled and migrate the keys.
• AccessDenied on the manifest but not the part. Writing the locked manifest needs the same s3:PutObject plus s3:PutObjectRetention grants as the data part. A policy that scopes retention rights to *.parquet will block the .manifest.json sibling; widen the resource pattern to the whole prefix.
• A batch spans midnight and lands in two partitions. build_partition deliberately rejects multi-day batches so a part never straddles a calendar boundary the reader can’t see. Split the buffer on UTC date upstream, or flush on a schedule aligned to midnight UTC rather than local time.
• Small-file sprawl slows the reader. High-frequency flushing produces thousands of tiny parts and turns a window read into a listing storm. Buffer to a target part size (roughly 64–256 MB) before committing; the content-addressed key still keeps the write idempotent.
• Hash matches but the row count drifts. That points at a schema-inference bug upstream, not corruption — a null-only column changed the Arrow types between batches. Pin AUDIT_SCHEMA explicitly, as the writer does, so every part is byte-comparable and column types never wander.

Related

• Audit Trail & Data Lineage Storage Patterns — parent section and the overall lineage contract
• Verifying Record Integrity with Cryptographic Checksums — deepening the per-part hash into a verifiable chain
• Building Append-Only Audit Logs in PostgreSQL — the row-level counterpart for hot, queryable history
• Security Boundary Design — least-privilege credentials and the write-path isolation this store assumes