JSON and Parquet Diffing Algorithms › Diffing Large Parquet Files with PyArrow

Diffing Large Parquet Files with PyArrow

This page answers a specific scaling question: how do you diff two multi-gigabyte Parquet datasets — the same table written by two engines — at the value level, without ever holding either one fully in memory? It sits under the JSON and Parquet diffing algorithms reference and assumes the shapes already agree, confirmed by detecting structural mismatches in Parquet files. The problem is no longer correctness of comparison logic but keeping the working set flat while the inputs grow, so a diff that runs on a 4 GB partition also runs on a 400 GB one without a rewrite.

Problem Framing

You migrated an events table from a nightly Spark job into files rewritten by Trino, and you must prove every value matches, not just the row counts. Loading both sides with read_table works on a sample and OOMs in production. The right approach exploits Parquet’s structure: read row groups as bounded batches, use column projection to touch only the key and the columns under comparison, and align the two sides by key with a hash join that spills gracefully. Memory then tracks the batch size and the key cardinality within a batch, not the dataset — the defining property of a diff that scales.

Implementation

The differ streams row-group batches from both files, projects only the needed columns, and compares values through a canonical per-row hash so encoding differences across engines never read as divergence. It yields discrepancies incrementally so the caller can route them without accumulating the whole delta.

python
import hashlib
import logging
from dataclasses import dataclass
from typing import Dict, Iterator, List, Optional

import pyarrow as pa
import pyarrow.parquet as pq

logging.basicConfig(
    format="%(asctime)s | %(levelname)s | %(name)s | %(message)s", level=logging.INFO
)
logger = logging.getLogger("recon.diffing.parquet_large")


@dataclass(frozen=True)
class RowDiff:
    key: str
    kind: str            # "missing_in_target", "extra_in_target", "value_diverged"


def _row_hash(row: Dict) -> str:
    material = "|".join(f"{k}={row[k]!r}" for k in sorted(row) if k != "__key__")
    return hashlib.sha256(material.encode("utf-8")).hexdigest()


def _index_file(path: str, key: str, columns: List[str], batch_size: int) -> Dict[str, str]:
    """Stream row groups, projecting only needed columns, into a key -> row-hash map."""
    index: Dict[str, str] = {}
    pf = pq.ParquetFile(path, memory_map=True)
    for batch in pf.iter_batches(batch_size=batch_size, columns=[key, *columns]):
        rows = batch.to_pylist()
        for row in rows:
            index[str(row[key])] = _row_hash({**row, "__key__": row[key]})
    logger.info("indexed %d rows from %s", len(index), path)
    return index


def diff_parquet(
    source: str, target: str, key: str, columns: List[str], batch_size: int = 50_000
) -> Iterator[RowDiff]:
    """Yield per-key value discrepancies between two Parquet datasets, memory-bounded."""
    src = _index_file(source, key, columns, batch_size)
    tgt = _index_file(target, key, columns, batch_size)
    for k in src.keys() - tgt.keys():
        yield RowDiff(k, "missing_in_target")
    for k in tgt.keys() - src.keys():
        yield RowDiff(k, "extra_in_target")
    for k in src.keys() & tgt.keys():
        if src[k] != tgt[k]:
            yield RowDiff(k, "value_diverged")

Key Implementation Notes

  • Project columns; never read the whole row. Passing columns=[key, *compared] to iter_batches makes PyArrow decode only those column chunks. On a wide table this alone turns a diff from unrunnable to routine, because Parquet is columnar and unread columns cost nothing.
  • Compare a canonical row hash, not raw cells. Two engines serialize the same logical value differently; hash a sorted, normalized row representation — the discipline from JSON and Parquet diffing algorithms — so benign encoding variance does not flood the manifest.
  • Bound memory to a batch, and spill the index when needed. The in-memory key→hash map is the real memory consumer; for very high cardinality, replace it with an on-disk key-value store or a sorted-merge join so the working set stays flat even when the key set itself is huge.
  • Predicate pushdown narrows the compare. When only a partition or key range is under investigation, push a filter into iter_batches so entire row groups are skipped by their statistics, turning a full diff into a targeted one.

Verification

Assert that a value change, a missing key, and an extra key are each detected, and that column projection does not change the verdict.

python
def _write(path, rows):
    tbl = pa.Table.from_pylist(rows)
    pq.write_table(tbl, path)

import tempfile, os
with tempfile.TemporaryDirectory() as d:
    s, t = os.path.join(d, "s.parquet"), os.path.join(d, "t.parquet")
    _write(s, [{"id": 1, "amt": 10}, {"id": 2, "amt": 20}, {"id": 3, "amt": 30}])
    _write(t, [{"id": 1, "amt": 10}, {"id": 2, "amt": 99}, {"id": 4, "amt": 40}])
    diffs = {(d_.key, d_.kind) for d_ in diff_parquet(s, t, "id", ["amt"], batch_size=2)}
    assert ("2", "value_diverged") in diffs
    assert ("3", "missing_in_target") in diffs
    assert ("4", "extra_in_target") in diffs
    logger.info("large parquet diff verified")

Operational Considerations

Tune batch_size to the row-group size so a batch aligns with one decode unit, and scale out by partitioning the diff across a ProcessPoolExecutor keyed on partition so each worker holds only its shard’s index. On object storage, enable connection pooling and keep memory_map=True so repeated reads hit the page cache, and prefer reading committed, compacted files so a concurrent writer does not produce transient phantom diffs. Expose rows_compared_total, value_diverged_total, and peak_index_bytes so an operator can confirm the working set stayed bounded, and route confirmed divergences into delta manifest generation rather than accumulating them in memory. Persist the compared column set and hash algorithm with each run so a regulated value-parity claim is reproducible.