Structural Diffing & Sync Engines › JSON and Parquet Diffing Algorithms

# JSON and Parquet Diffing Algorithms

JSON and Parquet diffing is the format-aware core of the comparison stage: the workload that proves a row-oriented JSON export and a columnar Parquet table hold the same logical data, despite two serialization models that disagree on almost every physical detail. JSON preserves arbitrary key ordering, coerces types implicitly, and represents absence in three different ways (missing key, explicit null, empty string); Parquet enforces a declared schema, stores nulls in a validity bitmap, dictionary-encodes repeated values, and packs columns into independently compressed row groups. For data engineers, migration specialists, Python pipeline builders, and platform operations teams, comparing these two by raw bytes is guaranteed to produce false divergence. This reference sits inside the Structural Diffing & Sync Engines stage and specializes its canonical intermediate representation for the JSON-versus-Parquet case.

The reason this deserves its own workload — rather than a generic value comparison — is that the two formats fail to agree for reasons that carry no business meaning. A "123" string in JSON is an int32 in Parquet; a key omitted from a JSON document is a validity-bit null in a Parquet column; 1.0 and 1 are the same number but different tokens. A diffing engine that cannot normalize these away will drown operators in phantom discrepancies, while one that normalizes too aggressively will hide real truncation and precision loss. This page builds the middle path: a deterministic fast path that hashes canonicalized chunks, and a semantic slow path that explains the rows the fast path flags.

# Architectural Boundaries

This workload begins the moment two aligned inputs are available — a newline-delimited JSON stream and a Parquet file that are believed to hold the same logical rows in the same key order — and it ends when it has emitted a per-chunk verdict stream: hash_match, row_count_drift, or semantic_mismatch. It consumes canonicalized, key-sorted inputs whose serialization contract is fixed upstream by data equivalence modeling, plus the type map that a stable cross-platform schema mapping supplies. It produces a delta stream that downstream routing turns into a discrepancy manifest — it never mutates either side.

Three concerns stay isolated inside the boundary. Canonicalization reduces both formats to one deterministic byte representation per row and nothing else. The fast path compares chunk digests and decides only whether a chunk is provably equal. The slow path runs an explaining diff exclusively over the chunks the fast path could not clear. Keeping these decoupled is what lets the engine spend cryptographic-hash cost on every chunk but expensive semantic-diff cost on only the tiny fraction that actually diverges. Structural anomalies that are not row-value differences at all — a field that changed type across the whole table, an array promoted to a struct — are out of scope here and belong to the structural mismatch detection workload, which reads footer metadata rather than row payloads.

# Prerequisites

Confirm each item before wiring JSON-and-Parquet diffing into a reconciliation run. Every one removes a common source of phantom divergence or unbounded memory growth.

• Row order is aligned. Both sides are sorted on the same reconciliation key, so chunk N on the JSON side and chunk N on the Parquet side describe the same rows — this engine compares position-aligned chunks, not a full outer join.
• Schema map resolved. The JSON-key → Parquet-column type mapping is fixed by the cross-platform schema mapping reference, so numeric-string coercion and nullability are unambiguous.
• Null policy declared. A single rule decides whether a missing JSON key, an explicit JSON null, and a Parquet validity-bit null collapse to one sentinel — set it before hashing, not per row.
• JSON is newline-delimited. The source emits one JSON object per line (NDJSON), so the reader can stream it in bounded chunks rather than parsing a multi-gigabyte array into memory.
• Dependency libraries pinned. pyarrow, xxhash, and deepdiff are version-pinned so digest bytes and diff output are reproducible across worker hosts.
• Tolerance profile chosen. Float epsilon and timestamp resolution come from the threshold tuning for tolerance reference, so the semantic path suppresses meaningless rounding without hiding real drift.

# Step-by-Step Implementation

The steps below build a chunked, schema-aware diff engine. Each ends with an assertion or an observable output so it can be verified in isolation before the next layer is added.

# Step 1 — Canonicalize a single row to deterministic bytes

Both formats must funnel through one serializer so that logically equal rows produce byte-identical output. Sort keys, coerce numeric strings to their schema type, drop null-valued keys to a single representation, and serialize with fixed separators.

import json
import logging
from typing import Any, Dict

logger = logging.getLogger("recon.jsonparquet")


def canonicalize_row(row: Dict[str, Any]) -> bytes:
    """Reduce a row to deterministic bytes, identical for JSON and Parquet sources.

    Sorted keys, coerced numeric strings, and dropped nulls make '1', 1, and 1.0
    collapse to one representation — the recurring root cause of phantom diffs.
    """
    def _coerce(val: Any) -> Any:
        if isinstance(val, str):
            try:
                return int(val) if "." not in val else float(val)
            except ValueError:
                return val
        return val

    sorted_items = sorted(
        (k, _coerce(v)) for k, v in row.items() if v is not None
    )
    return json.dumps(
        sorted_items, default=str, separators=(",", ":"), ensure_ascii=False
    ).encode("utf-8")

Verify that field order and a numeric-string mismatch normalize away:

a = canonicalize_row({"amount": "123", "id": 1, "note": None})
b = canonicalize_row({"id": 1, "amount": 123})
assert a == b   # key order, numeric coercion, and dropped null all reconcile

The same serialization contract that governs cross-engine digesting lives in the column-level checksum generation reference; this stage reuses it so that a hash computed here matches a hash computed upstream.

# Step 2 — Hash an aligned chunk from each side

Reduce a whole chunk to one digest by concatenating its canonical rows and hashing once. Both sides must use the same per-row canonicalization, or the fast path could never match. xxhash is a fast non-cryptographic digest chosen for the equality fast path — where the threat model is accidental divergence, not adversarial collisions.

import xxhash
import pyarrow as pa
from typing import List


def hash_rows(rows: List[Dict[str, Any]]) -> bytes:
    """Deterministic digest over a chunk's canonical rows."""
    return xxhash.xxh3_64(
        b"".join(canonicalize_row(r) for r in rows)
    ).digest()


def hash_parquet_batch(batch: pa.RecordBatch) -> bytes:
    """Same canonicalization as the JSON side, applied to a Parquet batch."""
    return hash_rows(batch.to_pylist())

Verify that a JSON chunk and the equivalent Parquet batch agree:

rows = [{"id": 1, "amount": "10"}, {"id": 2, "amount": "20"}]
batch = pa.RecordBatch.from_pylist([{"id": 1, "amount": 10}, {"id": 2, "amount": 20}])
assert hash_rows(rows) == hash_parquet_batch(batch)

# Step 3 — Stream both sides in aligned chunks and compare

Iterate Parquet in fixed-size record batches and pull the same number of NDJSON lines per batch. Compare row counts first — a mismatch is decisive and skips the hash entirely — then compare digests. Only when digests diverge does the engine fall through to the semantic path.

from pathlib import Path
from typing import Iterator
import pyarrow.parquet as pq
from deepdiff import DeepDiff


class ReconciliationError(Exception):
    """Raised when divergence exceeds the configured tolerance."""


def chunked_reconciliation(
    json_path: Path,
    parquet_path: Path,
    chunk_size: int = 25_000,
    max_mismatches: int = 100,
) -> Iterator[Dict[str, Any]]:
    """Stream JSON and Parquet in aligned chunks; hash fast-path, DeepDiff fallback.

    The JSON file is expected to be newline-delimited (one object per line).
    Yields a verdict record per chunk; raises once cumulative mismatches exceed
    max_mismatches so an unbounded divergence cannot run to completion.
    """
    parquet_file = pq.ParquetFile(parquet_path)
    mismatch_count = 0
    chunk_idx = 0

    try:
        with open(json_path, "r", encoding="utf-8") as jf:
            for pq_batch in parquet_file.iter_batches(batch_size=chunk_size):
                chunk_idx += 1
                json_rows: List[Dict[str, Any]] = []
                for _ in range(chunk_size):
                    line = jf.readline()
                    if not line:
                        break
                    json_rows.append(json.loads(line.strip()))

                if len(json_rows) != pq_batch.num_rows:
                    logger.warning(
                        "Row count drift at chunk %d: JSON=%d Parquet=%d",
                        chunk_idx, len(json_rows), pq_batch.num_rows,
                    )
                    yield {"chunk": chunk_idx, "status": "row_count_drift"}
                    continue

                if hash_rows(json_rows) == hash_parquet_batch(pq_batch):
                    yield {"chunk": chunk_idx, "status": "hash_match"}
                    continue

                # Slow path: explain only the rows the fast path could not clear.
                diffs = []
                for i, (j_row, p_row) in enumerate(
                    zip(json_rows, pq_batch.to_pylist())
                ):
                    diff = DeepDiff(j_row, p_row, ignore_order=True, verbose_level=2)
                    if diff:
                        diffs.append({"row_index": i, "diff": diff.to_dict()})

                mismatch_count += len(diffs)
                if mismatch_count > max_mismatches:
                    raise ReconciliationError(
                        f"Exceeded max_mismatches ({max_mismatches}) at chunk {chunk_idx}"
                    )
                yield {
                    "chunk": chunk_idx,
                    "status": "semantic_mismatch",
                    "details": diffs[:5],  # truncate for payload safety
                }

    except FileNotFoundError as exc:
        logger.error("Input file missing: %s", exc)
        raise ReconciliationError("Source file not found") from exc
    except json.JSONDecodeError as exc:
        logger.error("Malformed JSON at offset %d", exc.pos)
        raise ReconciliationError("Invalid JSON payload") from exc
    except pa.ArrowInvalid as exc:
        logger.error("Parquet read error: %s", exc)
        raise ReconciliationError("Parquet corruption or schema mismatch") from exc
    finally:
        logger.info("Reconciliation finished after %d chunks", chunk_idx)

Verify the fast path clears identical inputs and the slow path catches a planted divergence:

verdicts = list(chunked_reconciliation(Path("export.ndjson"), Path("table.parquet")))
assert all(v["status"] == "hash_match" for v in verdicts)   # clean run: every chunk matched

The verdict flow the two paths implement is summarized below; a mismatch on either the count check or the digest check routes the chunk to progressively more expensive handling.

When cumulative divergence crosses max_mismatches, the tiered degradation that decides whether to quarantine, resample, or halt is owned by the fallback chain implementation reference rather than being hard-coded here.

# Hash Algorithm Trade-offs

The fast path’s digest choice is a deliberate trade-off between throughput and the strength of the integrity claim the digest can support. A non-cryptographic hash is ideal for an internal equality short-circuit; a regulated audit trail that must defend against deliberate tampering needs a cryptographic one. Size the choice to the trust boundary the digest crosses.

For the position-aligned equality short-circuit on this page, xxhash.xxh3_64 is the default: the two sides are non-adversarial copies of the same pipeline output, so accidental-collision resistance is sufficient and throughput dominates. The moment a chunk digest is written into a discrepancy manifest that leaves the trust boundary, promote it to a cryptographic digest from Python’s hashlib — the hashlib documentation covers the SHA-2 and BLAKE2 constructors — so the recorded verdict is defensible. Never use MD5 for either role; its collision weakness disqualifies it from a credible audit trail. The policy governing which columns must carry an audit-grade digest is set by the security boundaries for reconciliation reference.

# Scaling and Performance

The engine is CPU-bound on canonicalization and hashing and I/O-bound on Parquet reads, so both dimensions must be engineered together.

Partition-aligned reads. Align chunk boundaries to Parquet row groups so each iter_batches call reads one compressed unit and touches only the columns under comparison. Column projection — reading the digest columns, not the whole schema — cuts both object-storage bytes fetched and decode cost, and co-locating diff workers with storage nodes keeps range requests local.

Batch sizing. The chunk_size controls the memory ceiling: worst-case residency is roughly chunk_size × avg_row_bytes on each side plus the two digest buffers. Start at 25k–50k rows and tune against observed RSS; wide blob columns should be sized by byte budget rather than row count, and columns that do not participate in the comparison should be excluded from the digest set entirely.

Memory bounding. Never materialize either side in full. The NDJSON reader must pull exactly chunk_size lines per Parquet batch so the two streams advance in lockstep — a buffered json.load over a multi-gigabyte array reintroduces the OOM failure mode this design exists to prevent. When extraction outruns comparison, the backpressure supplied by the upstream async batching for large datasets stage keeps in-flight rows bounded.

GIL and parallelism. xxhash and hashlib release the GIL during digest computation, so sharding partitions across a thread pool yields near-linear hashing throughput. The Python-level canonicalization loop, however, is GIL-bound; when serialization dominates — many small fields per row — shard by key range across a process pool instead, so each worker owns a disjoint, independently checkpointable partition. Pre-loading column metadata and dictionary encodings into a broadcast cache removes repeated schema-resolution and cold-fetch latency from the hot path.

# Failure Modes and Diagnostic Runbook

Each named failure mode lists its cause, the signal that detects it, and the remediation.

• Phantom diffs from unnormalized nulls. Cause: a missing JSON key, an explicit null, and a Parquet validity-bit null are not collapsed to one representation. Signal: semantic_mismatch on nearly every chunk, with DeepDiff reporting dictionary_item_added / removed on nullable fields. Remediation: funnel every value through canonicalize_row and fix the null sentinel before hashing; assert that two rows differing only in an absent key hash identically.
• Systematic float divergence. Cause: FLOAT32 Parquet storage versus double-precision JSON, or decimal-to-float widening, produces last-bit differences. Signal: semantic_mismatch isolated to numeric columns with tiny deltas. Remediation: apply a relative epsilon from the threshold tuning for tolerance profile in the semantic pass rather than treating exact inequality as drift.
• Misaligned chunk boundaries. Cause: the JSON and Parquet streams are not sorted on the same key, so position-aligned chunks describe different rows. Signal: row_count_drift clears but nearly every aligned chunk reports semantic_mismatch. Remediation: enforce an explicit primary-key sort on both sides upstream; this engine assumes alignment and does not join.
• OOM on a buffered JSON read. Cause: the source is a single JSON array parsed with json.load instead of streamed NDJSON. Signal: worker RSS climbs linearly to the OOM killer, independent of chunk_size. Remediation: require newline-delimited input and read exactly chunk_size lines per batch.
• Timestamp resolution mismatch. Cause: JSON millisecond strings versus Parquet microsecond TIMESTAMP values serialize differently. Signal: every timestamped row diverges by a constant sub-second amount. Remediation: normalize both to a common epoch resolution in canonicalization and compare within the declared temporal tolerance.
• Unbounded semantic-diff cost. Cause: a schema-wide type change makes every chunk fall through to DeepDiff, and max_mismatches is set too high. Signal: runtime explodes and the run stalls in the slow path. Remediation: keep max_mismatches tight so a systemic break raises ReconciliationError early, then route the whole table to structural mismatch detection instead of diffing row by row.

# In This Reference

This diffing workload has a dedicated companion that goes deeper on tool selection for the semantic slow path:

• Comparing JSON structures with Python diff libraries — benchmarks DeepDiff, jsondiff, and hand-rolled recursive comparison across nested payloads, weighing traversal cost, output granularity, and ignore-rule expressiveness for the fallback path built above.

# Frequently Asked Questions

# Related

• Structural Diffing & Sync Engines — the parent stage whose canonical intermediate representation this format-aware diff specializes.
• Structural mismatch detection — schema-level drift detection for table-wide type and layout changes that row diffing should not handle.
• Threshold tuning for tolerance — how to set numeric epsilons and temporal precision so the semantic pass suppresses rounding without hiding truncation.
• Fallback chain implementation — the tiered degradation strategy invoked when divergence exceeds tolerance.
• Comparing JSON structures with Python diff libraries — library selection for the semantic slow path.