Parquet Telemetry Schema Design

A caption QC pipeline that appends violation telemetry as loosely-typed rows becomes unqueryable within a quarter: one day the measured column infers as int64, the next as double, and a warehouse scan across the two files throws a type-unification error mid-audit. The fix is an explicit columnar schema pinned before the first write — every field typed, every future column nullable, partitioned so a query for “all FCC 47 CFR § 79.1 reading-rate breaches in the last 30 days” reads two directories instead of a year of files. This page defines that schema with pyarrow: one row per QC violation carrying asset_id, cue_tc, rule, clause, severity, measured, threshold, framerate, territory and a UTC run_ts, written as Hive-partitioned, dictionary-encoded, zstd-compressed Parquet that the scheduled QC reporting job emits on every run.

The schema and partitioned writer

# qc_parquet_schema.py — stable, partitioned Parquet telemetry for caption QC violations
from __future__ import annotations

import datetime as dt
from dataclasses import dataclass

import pyarrow as pa
import pyarrow.dataset as ds

# One row per violation. Types are pinned so every daily write is byte-compatible with
# the last — a warehouse scan never has to reconcile int64-vs-double column drift.
VIOLATION_SCHEMA = pa.schema(
    [
        pa.field("asset_id", pa.string(), nullable=False),                       # program/asset UUID
        pa.field("cue_tc", pa.string(), nullable=False),                         # SMPTE ST 12-1 HH:MM:SS:FF
        pa.field("rule", pa.string(), nullable=False),                           # partition key (plain string)
        pa.field("clause", pa.dictionary(pa.int16(), pa.string()), nullable=False),   # FCC 47 CFR § 79.1 ...
        pa.field("severity", pa.dictionary(pa.int8(), pa.string()), nullable=False),  # info | warning | error
        pa.field("measured", pa.float64(), nullable=False),                      # observed value, e.g. 24.3 cps
        pa.field("threshold", pa.float64(), nullable=False),                     # ceiling breached, e.g. 20.0 cps
        pa.field("framerate", pa.float64(), nullable=False),                     # 29.97 | 25.0 | 23.976
        pa.field("territory", pa.dictionary(pa.int8(), pa.string()), nullable=False),  # US | GB | CA
        pa.field("run_ts", pa.timestamp("us", tz="UTC"), nullable=False),        # report run instant, UTC
        # Additive evolution: every NEW column must be nullable so older files still read.
        pa.field("detector_version", pa.string(), nullable=True),
    ],
    metadata={b"schema_version": b"1.3.0", b"owner": b"qc-telemetry"},
)


@dataclass(frozen=True)
class Violation:
    asset_id: str
    cue_tc: str
    rule: str
    clause: str
    severity: str
    measured: float
    threshold: float
    framerate: float
    territory: str
    run_ts: dt.datetime
    detector_version: str | None = None


def to_table(rows: list[Violation]) -> pa.Table:
    """Bind rows to the schema — a type mismatch raises here, not months later at read time."""
    cols = {name: [getattr(r, name) for r in rows] for name in VIOLATION_SCHEMA.names}
    table = pa.table(cols, schema=VIOLATION_SCHEMA)  # dict fields encode from plain python strings
    # Derive the date partition key from run_ts so partitions align to the report window,
    # not to wall-clock write time (a run at 00:03 UTC still lands in the prior day's window).
    run_date = [r.run_ts.astimezone(dt.timezone.utc).date().isoformat() for r in rows]
    return table.append_column("run_date", pa.array(run_date, pa.string()))


def write_partitioned(rows: list[Violation], base_dir: str) -> None:
    """Write base_dir/run_date=YYYY-MM-DD/rule=<rule>/part-*.parquet, idempotent per window."""
    table = to_table(rows)
    part = ds.partitioning(
        pa.schema([("run_date", pa.string()), ("rule", pa.string())]),
        flavor="hive",                              # run_date=.../rule=... directory layout
    )
    fmt = ds.ParquetFileFormat()
    opts = fmt.make_write_options(
        compression="zstd",                         # 2-4x over snappy on repetitive telemetry
        compression_level=6,
        use_dictionary=True,                        # Parquet-level dict-encode of repeated strings
        write_statistics=True,                      # min/max stats drive predicate pushdown
    )
    ds.write_dataset(
        table,
        base_dir=base_dir,
        format=fmt,
        file_options=opts,
        partitioning=part,
        max_rows_per_group=128_000,                 # target row-group size for scan efficiency
        min_rows_per_group=64_000,                  # coalesce small groups; fight the small-file problem
        existing_data_behavior="overwrite_or_ignore",  # re-running one window replaces, never doubles
        basename_template="part-{i}.parquet",
    )


if __name__ == "__main__":
    now = dt.datetime.now(dt.timezone.utc)
    sample = [
        Violation("a1b2", "01:00:12:04", "reading_rate", "FCC 47 CFR § 79.1", "error",
                  24.3, 20.0, 29.97, "US", now, "qc-detect/4.2.0"),
        Violation("a1b2", "01:03:44:11", "line_length", "CEA-608 32-column", "warning",
                  34.0, 32.0, 29.97, "US", now, "qc-detect/4.2.0"),
    ]
    write_partitioned(sample, "qc_telemetry")

Code walkthrough

The whole design rests on pa.schema([...]) being written by hand rather than inferred. pyarrow will happily guess types from the first batch it sees, but inference is per-file: a window with only integer measured values produces an int64 column, and the next window with a 24.3 produces a double, so a cross-day dataset scan fails to unify the two. Declaring measured and threshold as pa.float64() once fixes the physical type for every write for the life of the table. run_ts is a timezone-aware pa.timestamp("us", tz="UTC") — microsecond precision is ample for a report instant, and pinning the zone to UTC means a query never has to reason about a naive timestamp that might be Los Angeles or London.

clause, severity and territory are pa.dictionary types because their cardinality is tiny and fixed — a handful of regulatory clauses, three severities, a short list of markets. A dictionary column stores each distinct value once and indexes into it, so a million-row file holding the string "FCC 47 CFR § 79.1" a million times costs one copy of the string plus a column of small integers. rule is deliberately a plain pa.string() instead: it is a partition column, and partition values are lifted out of the file into the directory path, so encoding it as a dictionary inside the file would be wasted work and can confuse the partitioning schema, which expects a plain string.

to_table binds the rows to the schema explicitly. Passing schema=VIOLATION_SCHEMA to pa.table converts each Python list to the declared Arrow type — the plain strings destined for clause/severity/territory become dictionary arrays, and any value that cannot convert raises immediately, at write time, where a stack trace points at the bug. It then derives run_date from run_ts rather than from datetime.now(): aligning the partition to the report window (the same half-open window the daily QC report job computes) keeps a job that fires at 00:03 UTC writing into the previous day’s directory instead of splitting one window across two partitions.

write_partitioned uses pyarrow.dataset.write_dataset — the modern replacement for the legacy write_to_dataset — with a Hive partitioning on run_date then rule. The layout run_date=2026-07-16/rule=reading_rate/part-0.parquet lets a query engine prune whole directories from a WHERE run_date = '...' AND rule = '...' predicate before opening a single file. use_dictionary=True and write_statistics=True in the write options turn on Parquet-level dictionary encoding and min/max column statistics, so even non-partition predicates (measured > 30) skip row groups whose statistics exclude a match. max_rows_per_group=128_000 sets the row-group granularity that predicate pushdown operates at, and existing_data_behavior="overwrite_or_ignore" makes a re-run of the same window replace its partition rather than append a duplicate copy — the idempotency the parent scheduler relies on.

Schema & partition reference

The clause and threshold values above are the same ones the validators upstream enforce; their authoritative source is the scheduled QC reporting threshold table and the underlying character-rate limits in QC, and the drift figures come from automated sync drift detection. The 20 cps ceiling that populates threshold traces to 47 CFR § 79.1 readability expectations.

Edge cases & known gotchas

• Schema drift is the primary failure. Never let a new detector silently add a column mid-stream; a reader opening the dataset with the old schema either drops it or errors. Bump schema_version in the metadata, add the field as nullable=True, and only ever append — renaming or retyping a column breaks every historical partition.
• Null handling must be explicit. A field the detector cannot always populate (detector_version on legacy rows) must be declared nullable=True; a nullable=False column that receives a None raises at to_table, which is correct for measured but wrong for genuinely optional metadata.
• Timezone in run_ts. Storing a naive timestamp is a latent bug: a query that filters run_ts against a UTC bound silently mis-selects rows written by a machine in another zone. Pin tz="UTC" in the type and normalize with astimezone(timezone.utc) before the value ever reaches Arrow.
• The small-file problem. A job that writes one tiny Parquet file per asset per rule per day produces millions of sub-megabyte files whose metadata overhead dwarfs their data and cripples scan planning. Batch a whole window into one write_dataset call and set min_rows_per_group so row groups coalesce; compact stragglers on a slower cadence.
• Dictionary blow-up on the wrong column. Dictionary-encoding a high-cardinality field like asset_id inflates the file with a dictionary nearly as large as the data. Reserve pa.dictionary for genuinely low-cardinality fields; leave identifiers as plain strings.

Integration hook

This schema is the columnar substrate under one branch of the scheduled QC reporting fork: where the canonical JSON envelope feeds the build gate, this partitioned Parquet copy feeds the warehouse and every trend query an auditor runs against it. The rows are produced by the aggregation core in generating daily QC reports with Python, and the same violation records are projected into the human-facing contract described in compliance dashboard JSON output format. Because the writer is idempotent per window, the cron-triggered batch runner can safely re-drive any date range without producing duplicate telemetry.

Related

• Scheduled QC report generation — parent reference: the scheduler trigger, scoring, and dual JSON/Parquet output this schema serves.
• Generating daily QC reports with Python — the aggregation core that emits the violation rows written here.
• Compliance dashboard JSON output format — the versioned JSON projection of the same telemetry for the dashboard.
• Automated sync drift detection — one source of the measured/threshold drift rows this schema stores.

Part of: Automated QC Validation & Reporting — the deterministic caption QC and reporting reference.