DuckDB vs Polars for Royalty Aggregation at Scale

Once monthly DSP sales CSVs have been parsed and typed, the next bottleneck is aggregation: grouping hundreds of millions of stream-count rows by isrc, territory_code, and reporting_period, then joining the result against a rights-share table to compute payee payouts. Two tools now dominate this stage of Automated CSV Parsing for Sales Reports: DuckDB, an embedded columnar SQL engine, and Polars (>=0.20), a Rust-backed dataframe library with a lazy query API. Both beat pandas at this scale, but they diverge in how they handle out-of-core spilling, join strategy, and how comfortably a royalty team can audit the transformation logic after the fact.

Aggregation Trade-offs

Prerequisites & Assumptions

Examples assume Python 3.11+, duckdb>=0.10, polars>=0.20, and pyarrow>=14 for interchange, with royalty CSVs already normalized to Parquet by an upstream step such as optimizing pandas for 10GB royalty CSVs. The working dataset is a monthly sales fact table (isrc, territory_code, reporting_period, stream_count, gross_payout) joined against a rights_shares table (isrc, payee_id, right_share_pct). Both engines below run on a single machine with 16–32GB RAM against files in the 5–50GB range — the size band where pandas reliably fails but a distributed cluster is still overkill.

Step 1: Group-by aggregation in DuckDB

DuckDB queries Parquet files directly without an explicit load step, and its query planner pushes the GROUP BY down to the scan.

import duckdb

con = duckdb.connect(database=":memory:")
con.execute("SET memory_limit='24GB'")
con.execute("SET temp_directory='/var/tmp/duckdb_spill'")

result = con.execute("""
    SELECT
        isrc,
        territory_code,
        reporting_period,
        SUM(stream_count) AS total_streams,
        SUM(gross_payout)::DECIMAL(18,4) AS total_gross_payout
    FROM read_parquet('sales/2024-q3/*.parquet')
    GROUP BY isrc, territory_code, reporting_period
""").arrow()

Setting temp_directory explicitly is what lets DuckDB spill intermediate hash tables to disk once memory_limit is exceeded, rather than raising an out-of-memory error mid-aggregation — critical for a quarterly batch where row counts can double without warning.

Step 2: Equivalent aggregation in Polars

Polars expresses the same aggregation through its lazy API, deferring execution until .collect() so the query optimizer can fuse the scan, filter, and group-by into one pass.

import polars as pl

lazy_sales = pl.scan_parquet("sales/2024-q3/*.parquet")

result = (
    lazy_sales
    .group_by(["isrc", "territory_code", "reporting_period"])
    .agg([
        pl.col("stream_count").sum().alias("total_streams"),
        pl.col("gross_payout").sum().cast(pl.Decimal(18, 4)).alias("total_gross_payout"),
    ])
    .collect(streaming=True)
)

streaming=True enables Polars’ out-of-core execution engine, processing the query in batches rather than materializing the full intermediate hash table — but skewed keys (a single viral track dominating one isrc group) can still force a large partial aggregate into memory at once.

Step 3: Joining aggregated sales to the rights table

The payout-critical step is joining aggregated streams to rights_shares and multiplying by right_share_pct. Both engines support this, but DuckDB’s SQL join reads more directly as an auditable formula.

# DuckDB: join and compute payee-level payout in one statement
payout = con.execute("""
    SELECT
        s.isrc,
        s.territory_code,
        r.payee_id,
        s.total_gross_payout * (r.right_share_pct / 100.0) AS payee_payout
    FROM aggregated_sales s
    JOIN rights_shares r ON s.isrc = r.isrc
""").arrow()

# Polars: equivalent join expressed as a lazy chain
payout = (
    aggregated_sales_lazy
    .join(rights_shares_lazy, on="isrc", how="inner")
    .with_columns(
        (pl.col("total_gross_payout") * (pl.col("right_share_pct") / 100.0))
        .alias("payee_payout")
    )
    .collect(streaming=True)
)

Step 4: Choosing based on downstream consumers

If royalty managers or auditors need to inspect or modify the aggregation logic directly, DuckDB’s SQL is the lower-friction choice — it can also be pointed at the same Parquet files from a BI tool without any Python in between. If the pipeline is entirely Python-native and already leans on Polars elsewhere (for example alongside optimizing pandas for 10GB royalty CSVs migrations), staying in the lazy dataframe API avoids a second query language in the codebase.

Verification & Validation

Regardless of engine, validate the aggregation with a split-sum assertion before writing to the ledger: for every isrc, the sum of payee_payout across all payee_id rows should equal total_gross_payout within a cent of rounding tolerance.

tolerance = 0.01
check = payout.group_by("isrc").agg(pl.col("payee_payout").sum().alias("summed"))
# join back to total_gross_payout and assert abs(summed - total_gross_payout) < tolerance

Log the row count in, row count out, and total gross_payout before and after aggregation to the append-only audit trail, keyed by batch_id. A mismatch between pre- and post-aggregation totals — even a fractional-cent difference beyond Decimal rounding — should halt the batch rather than proceed to payout generation.

Edge Cases & Gotchas

Territory-code mismatches are the most common silent failure: a sales file using "GB" joined against a rights table keyed on "UK" produces an inner join that drops rows without raising an error in either engine — always assert row-count parity between the group-by output and the post-join output. DuckDB’s DECIMAL casts fail loudly on values exceeding the declared precision (e.g., a corrupted gross_payout of 1e20), whereas Polars’ Decimal cast can silently produce null in some versions — pin polars>=0.20.19 where this was hardened, and always check .null_count() on cast columns before proceeding. Orphaned ISRCs with no matching row in rights_shares disappear under an inner join in both engines; use a LEFT JOIN (DuckDB) or how="left" (Polars) and route unmatched rows to a fallback royalty queue instead of dropping them. Finally, DuckDB’s memory_limit and temp_directory settings are connection-scoped, not global — forgetting to set them on a freshly opened connection in a worker pool silently reverts to default limits under concurrent load.

Related

Both engines slot in after the CSV-to-Parquet normalization covered in optimizing pandas for 10GB royalty CSVs, and the choice between them feeds directly into the reconciliation gates defined across Automated CSV Parsing for Sales Reports.