Partitioning Royalty Data by Territory and Period

Every royalty lake eventually hits the same wall: a query for one label’s Q3 payouts in Germany scans terabytes of Parquet because the underlying files carry no structural hint about what’s inside them. Choosing a Hive-style layout keyed on reporting_period and territory_code is the single decision inside the Data Lake Architecture for Streaming Metrics that determines whether a reconciliation job finishes in seconds or hours, and whether a rights holder audit can isolate one jurisdiction’s numbers without touching the rest of the catalog. Get the partition grain wrong and you either drown the object store in small files or force every downstream engine to read data it will immediately discard.

Partitioning Scheme Trade-offs

The partition key order and grain directly change how much data a query engine like DuckDB or Spark must open before it can apply a predicate, and how easily a compaction job can keep file counts sane. The table below assumes a mid-size catalog: roughly 40 territories, monthly reporting_period values, and 200–800 million streaming events per month.

For most royalty lakes, reporting_period as the outer key and territory_code as the inner key wins because the dominant access pattern is “close this month, then break it down by territory for remittance.” Confirm the choice against your own query logs with DuckDB vs. Polars for Royalty Aggregation before committing to a physical layout, since re-partitioning a multi-terabyte lake later is expensive.

Prerequisites & Assumptions

This walkthrough assumes Python 3.11+, pyarrow>=14, duckdb>=0.10, and royalty events that have already passed field-level validation — territory codes should already be ISO 3166-1 alpha-2 and reporting periods normalized to YYYY-MM before they reach the partition writer, ideally enforced upstream by Schema Validation with Pydantic. Records are assumed to arrive as a stream of validated dictionaries or a pyarrow.Table, not raw DSP JSON.

Implementation

Step 1: Write Hive-Partitioned Parquet with an Explicit Schema

Use pyarrow.dataset.write_dataset rather than a manual os.makedirs loop — it handles directory creation, partition value encoding, and file naming consistently.

import pyarrow as pa
import pyarrow.dataset as ds

def write_royalty_partitions(table: pa.Table, root_path: str) -> None:
    """
    Writes a validated royalty batch into a Hive-style layout:
    root/reporting_period=2025-06/territory_code=DE/part-*.parquet
    """
    ds.write_dataset(
        data=table,
        base_dir=root_path,
        format="parquet",
        partitioning=ds.partitioning(
            pa.schema([
                ("reporting_period", pa.string()),
                ("territory_code", pa.string()),
            ]),
            flavor="hive",
        ),
        existing_data_behavior="overwrite_or_ignore",
        max_rows_per_file=1_000_000,
        min_rows_per_group=100_000,
    )

max_rows_per_file caps individual file size so a single territory’s monthly volume doesn’t produce one unwieldy file, while min_rows_per_group prevents the opposite failure mode of hundreds of tiny row groups per file.

Step 2: Prune Partitions at Query Time

A correctly partitioned lake is only useful if the query layer actually skips irrelevant directories. Filter on the partition columns explicitly rather than relying on the engine to infer pruning from a broader predicate.

import duckdb

def query_period_territory(root_path: str, reporting_period: str, territory_code: str):
    con = duckdb.connect()
    return con.execute(
        """
        SELECT isrc, payee_id, right_share_pct, SUM(net_payout_usd) AS total_payout
        FROM read_parquet(?, hive_partitioning=1)
        WHERE reporting_period = ? AND territory_code = ?
        GROUP BY isrc, payee_id, right_share_pct
        """,
        [f"{root_path}/**/*.parquet", reporting_period, territory_code],
    ).arrow()

hive_partitioning=1 tells DuckDB to read reporting_period and territory_code from directory names instead of decoding them from file contents, so the filter eliminates whole directories before a single byte of Parquet is decompressed.

Step 3: Compact Small Files After Each Period Close

Frequent micro-batch writes from DSP polling jobs fragment each territory partition into dozens of sub-megabyte files. Run a compaction pass once a reporting_period partition is closed for new writes.

import pyarrow.dataset as ds
import pyarrow.parquet as pq

def compact_partition(partition_path: str, target_rows_per_file: int = 1_000_000) -> None:
    """
    Reads every fragment in a closed partition and rewrites it as a small
    number of right-sized files, then the caller swaps the directory atomically.
    """
    dataset = ds.dataset(partition_path, format="parquet")
    scanner = dataset.scanner(batch_size=target_rows_per_file)
    for i, batch in enumerate(scanner.to_batches()):
        pq.write_table(
            pa.Table.from_batches([batch]),
            f"{partition_path}/compacted-{i:04d}.parquet",
            compression="zstd",
        )

Compaction should be idempotent: write to a staging directory, verify row counts against the source partition, then atomically rename staging into place and delete the pre-compaction fragments — never delete before the rewrite is confirmed complete.

Verification & Validation

After writing or compacting a partition, assert that row counts and split-sum totals match the pre-write batch exactly:

def verify_partition_integrity(source_rows: int, source_payout_total: float,
                                 partition_path: str) -> None:
    con = duckdb.connect()
    row = con.execute(
        "SELECT COUNT(*), SUM(net_payout_usd) FROM read_parquet(?)",
        [f"{partition_path}/*.parquet"],
    ).fetchone()
    assert row[0] == source_rows, f"row count drift: {row[0]} != {source_rows}"
    assert abs(row[1] - source_payout_total) < 0.01, "payout total drift after write"

Log the assertion result — including the partition key and a batch checksum — to the append-only audit trail so a discrepancy in a later financial reconciliation can be traced back to the exact write or compaction job that introduced it.

Edge Cases & Gotchas

Late-arriving corrections for an already-closed reporting_period are the most common source of pain: writing directly into a compacted partition reintroduces the small-file problem, so route corrections into a reporting_period=2025-06/territory_code=DE/_corrections/ subdirectory and merge on the next compaction cycle rather than triggering an immediate rewrite. Territory code mismatches between DSP-reported values (UK) and ISO 3166-1 (GB) will silently create a duplicate, near-empty partition unless normalization happens before the partition writer runs — validate against the same territory enum used elsewhere in ingestion. NULL or unrecognized territory codes should route to an explicit territory_code=UNKNOWN partition rather than being dropped, so unresolved royalty rows stay auditable instead of vanishing from the lake. Finally, watch for schema evolution: adding a new column mid-stream without a default value will cause older partitions to have missing fields, which breaks naive SELECT * queries across period boundaries unless the read path tolerates schema union.

Related

Partition layout decisions only pay off once ingestion has quarantined bad rows before they’re written, which is the job of Dead-Letter Queue Patterns for Failed Royalty Records; together they keep the Data Lake Architecture for Streaming Metrics both queryable and financially trustworthy.