Expense Audit Guide Python · Pipelines · Compliance

Chunking large receipt batches for OCR throughput

A 40,000-receipt month-end upload OOM-kills a worker the instant it is read into a list, because peak memory scales with the size of the run rather than the size of one unit of work. This guide splits that upload into bounded, order-tagged chunks streamed through a process pool, so OCR throughput grows with cores while resident memory stays flat — the chunk-sizing and result-reassembly detail delegated by the parent Async Batch Processing guide, inside the broader Receipt Ingestion & OCR Data Extraction framework.

The goal is not raw speed alone. It is bounded, deterministic speed: a fixed memory ceiling regardless of how large a batch arrives, and downstream records emitted in the same order they were submitted so the audit ledger stays reconstructable.

Streaming chunker feeding a process pool with ordered result reassembly An unbounded upload batch of N receipts enters a chunk generator that yields fixed-size, sequence-numbered chunks (size 200) lazily. Chunks fan out to a process pool of four workers p0 to p3 that run CPU-bound OCR in parallel and complete out of order. Their results land in a reorder buffer, a min-heap keyed by sequence number, which releases a chunk downstream only when the next contiguous sequence is ready, producing an ordered stream into OCR parse and policy stages. A memory-profile band contrasts buffering the whole batch, whose resident memory grows with N and OOM-kills at month-end, against streaming chunks, whose resident memory is flat and tracks chunk_size times pool_width times average receipt bytes. Split → stream → parallelize → reassemble in order Peak memory tracks chunk_size × pool_width, never the batch size N. Upload batch N receipts Chunk generator yield · size=200 seq 0,1,2… p0 · OCR p1 · OCR p2 · OCR p3 · OCR process pool ×4 Reorder buffer min-heap by seq out of order in order Ordered stream → parse / policy Memory profile Buffer whole batch RSS ∝ N · OOM-killed at month-end surge Stream chunks RSS ∝ chunk_size × pool_width · flat under any N Peak resident memory ceiling RSS ≈ chunk_size × (pool_width + max_in_flight) × avg_receipt_bytes Size max_in_flight and chunk_size before raising pool_width.

Why Standard Approaches Fail

The obvious approaches to a very large batch each fail in a way that either exhausts memory or corrupts the audit trail:

  • List materialization. receipts = list(db.stream_receipts()) followed by a for loop reads every record — and often every decoded image buffer — into RAM before any OCR runs. Peak resident memory becomes a function of total volume, so a batch that is fine at 2,000 receipts OOM-kills the same fixed-size worker at 40,000. The break is discontinuous: it works in staging and dies in production month-end.
  • Naive Pool.map over the whole list. Handing the entire list to multiprocessing.Pool.map preserves order but still materializes all inputs and all results at once, so both ends of the pipe balloon. imap streams inputs but a slow chunk stalls the ordered iterator behind it, re-introducing head-of-line blocking on the exact faded scans that take longest to OCR — the ones tuned in Tesseract OCR configuration.
  • Discarding submission order. Firing every receipt at a pool and collecting results as they finish is memory-friendly but loses ordering. When downstream posting to the ledger is order-sensitive — sequential voucher numbers, first-seen-wins duplicate resolution — nondeterministic completion order produces a non-reproducible audit trail that internal-control reviews reject.

The remedy is a generator that yields fixed-size, sequence-numbered chunks; a bounded window of chunks in flight across a process pool; and a small reorder buffer that restores submission order on the way out.

Architecture & Algorithm

Chunking has three moving parts: a lazy generator that never holds more than one chunk of source records, a bounded submission window that caps how many chunks are in flight (this, not the pool size, is what bounds memory), and a reorder buffer that absorbs out-of-order completion and re-emits in sequence.

The generator consumes its source as an iterator and tags each chunk with a monotonic seq. Because it appends to a small buffer and clears it on every yield, resident memory holds chunk_size records, not the full batch.

from __future__ import annotations

import logging
from dataclasses import dataclass
from typing import Any, Iterable, Iterator

logger = logging.getLogger("expense.ingest.chunking")


@dataclass(frozen=True)
class ReceiptRef:
    receipt_id: str
    image_uri: str


@dataclass(frozen=True)
class Chunk:
    seq: int
    receipts: tuple[ReceiptRef, ...]


def stream_chunks(receipts: Iterable[ReceiptRef], chunk_size: int = 200) -> Iterator[Chunk]:
    """Yield fixed-size, sequence-numbered chunks lazily so RSS stays flat.

    The source is consumed as an iterator, so only chunk_size ReceiptRef
    objects are resident at once regardless of how many receipts arrive.
    """
    if chunk_size < 1:
        raise ValueError("chunk_size must be >= 1")
    buffer: list[ReceiptRef] = []
    seq = 0
    for receipt in receipts:
        buffer.append(receipt)
        if len(buffer) == chunk_size:
            yield Chunk(seq=seq, receipts=tuple(buffer))
            seq += 1
            buffer.clear()
    if buffer:                       # final short chunk
        yield Chunk(seq=seq, receipts=tuple(buffer))

The unit of CPU work is a top-level function so it is picklable into worker processes. It returns its seq alongside results, which is what makes ordered reassembly possible without a shared data structure.

def _run_ocr(image_uri: str) -> str:
    """Bind to your Tesseract wrapper; kept pure so chunks stay picklable."""
    return f"OCR::{image_uri}"


def ocr_chunk(chunk: Chunk) -> tuple[int, list[dict[str, Any]]]:
    """CPU-bound OCR for one chunk, executed in a worker process.

    Returns (seq, results) so the parent can restore submission order after
    workers complete out of order.
    """
    results = [
        {"receipt_id": r.receipt_id, "text": _run_ocr(r.image_uri)}
        for r in chunk.receipts
    ]
    return chunk.seq, results

The driver keeps at most max_in_flight chunks submitted at once. Each time a future completes it submits the next chunk, so the pool stays saturated without ever queuing the whole batch. A min-heap keyed by seq, plus a next_emit cursor, releases results only when the contiguous next chunk is ready — bounding the reorder buffer to at most the number of in-flight chunks.

import heapq
from concurrent.futures import FIRST_COMPLETED, Future, ProcessPoolExecutor, wait


def process_in_order(
    receipts: Iterable[ReceiptRef],
    *,
    chunk_size: int = 200,
    pool_width: int = 4,
    max_in_flight: int = 8,
) -> Iterator[tuple[int, list[dict[str, Any]]]]:
    """Process chunks in parallel but yield them in submission order.

    Completion order is nondeterministic, so results land in a small heap
    keyed by seq and are released only when the contiguous next seq is ready.
    Memory is bounded by max_in_flight chunks, never by the total batch.
    """
    chunks = stream_chunks(receipts, chunk_size=chunk_size)
    pending: dict[int, list[dict[str, Any]]] = {}
    ready: list[int] = []
    next_emit = 0

    with ProcessPoolExecutor(max_workers=pool_width) as pool:
        in_flight: set[Future] = set()
        for _ in range(max_in_flight):           # prime the submission window
            chunk = next(chunks, None)
            if chunk is None:
                break
            in_flight.add(pool.submit(ocr_chunk, chunk))

        while in_flight:
            done, in_flight = wait(in_flight, return_when=FIRST_COMPLETED)
            for fut in done:
                seq, results = fut.result()
                pending[seq] = results
                heapq.heappush(ready, seq)
                logger.info("chunk_done seq=%d size=%d", seq, len(results))
                nxt = next(chunks, None)         # refill to keep the pool busy
                if nxt is not None:
                    in_flight.add(pool.submit(ocr_chunk, nxt))
            while ready and ready[0] == next_emit:  # emit contiguous prefix
                s = heapq.heappop(ready)
                yield s, pending.pop(s)
                next_emit += 1

The invariant that matters for capacity planning: at any instant the process holds max_in_flight chunks of inputs plus at most max_in_flight chunks of buffered results, so peak memory is chunk_size × (pool_width + max_in_flight) × avg_receipt_bytes and is completely independent of the batch length.

Chunk size itself is a throughput/memory tradeoff, not a constant. Larger chunks amortize per-task pickling and dispatch overhead but raise peak RSS and lengthen the tail a single slow chunk holds. Derive it from the memory budget rather than guessing:

def recommended_chunk_size(
    mem_budget_mb: int,
    pool_width: int,
    avg_receipt_mb: float,
    max_in_flight: int = 8,
    floor: int = 25,
    ceil: int = 500,
) -> int:
    """Largest chunk_size whose worst-case resident set fits the budget."""
    if avg_receipt_mb <= 0:
        raise ValueError("avg_receipt_mb must be positive")
    slots = pool_width + max_in_flight
    raw = int(mem_budget_mb / (slots * avg_receipt_mb))
    return max(floor, min(ceil, raw))

Step-by-Step Integration

  1. Stream the source, never list it. Feed process_in_order a lazy iterator — a server-side cursor or yield-based reader — so the generator, not a materialized list, is the only thing holding records.

  2. Set chunk_size from the memory budget. Call recommended_chunk_size(mem_budget_mb, pool_width, avg_receipt_mb) against the container limit, then pin the result in config so a capacity change is reviewed, not silent.

  3. Match pool_width to physical cores, not max_in_flight. OCR is CPU-bound, so pool_width = os.cpu_count() maximizes parallelism; max_in_flight should exceed pool_width by a small margin (2×) so a worker never idles waiting for the driver to submit.

  4. Verify order and completeness before wiring to the ledger. Emitted sequence numbers must be strictly increasing and cover every chunk with none dropped or duplicated:

    refs = [ReceiptRef(f"R-{i}", f"s3://b/{i}.jpg") for i in range(1000)]
seqs = [seq for seq, _ in process_in_order(refs, chunk_size=200, pool_width=4)]
assert seqs == sorted(seqs)             # emitted in submission order
assert seqs == list(range(len(seqs)))   # contiguous: nothing dropped or repeated
assert len(seqs) == 5                    # ceil(1000 / 200)

  5. Confirm RSS is flat across batch sizes. Run 1,000 and 40,000 receipts with the same chunk_size and max_in_flight; peak RSS should be nearly identical. A rising curve means something upstream is still materializing the batch.

  6. Hand ordered results to the next stage. Pass the ordered stream into line-item parsing and policy validation; because order is preserved, sequential voucher assignment and first-seen duplicate resolution stay deterministic across re-runs.

Edge Cases & Gotchas

Edge condition What breaks Mitigation
One slow chunk (faded scans) Reorder buffer holds later chunks until the laggard finishes, growing memory Cap max_in_flight; keep chunks small so a stall blocks fewer records
Final short chunk dropped Off-by-one loses the last < chunk_size receipts Yield the trailing buffer after the loop (handled in stream_chunks)
Unpicklable chunk payload ProcessPoolExecutor.submit raises on non-picklable objects Keep chunks as plain dataclasses of URIs/ids; never embed open file handles
chunk_size too large Per-worker RSS spikes; a single failure re-does more work Derive size from the memory budget; prefer more, smaller chunks
Exception inside ocr_chunk fut.result() re-raises and aborts the whole run Catch per chunk inside the worker and return a partial/failed marker
Ordering assumed but not needed Reorder buffer adds latency for no benefit Drop the heap and yield on completion when downstream is order-agnostic
Pool re-spawned per batch Interpreter start-up cost dominates on many small batches Reuse one ProcessPoolExecutor across batches; the parent Async Batch Processing engine owns its lifecycle

FAQ

How do I choose a chunk size?

Derive it from the memory budget, not intuition. Worst-case resident memory is roughly chunk_size × (pool_width + max_in_flight) × avg_receipt_bytes, so solve that for the largest chunk_size that fits your container limit, then clamp it to a sane floor and ceiling. Start around 100–200 receipts, watch peak RSS and per-chunk latency, and change one variable at a time.

Why keep submission order at all — isn’t completing as fast as possible the point?

Throughput and ordering are independent. Chunks are processed fully in parallel; the reorder buffer only controls the sequence in which finished results are released. You need that order when downstream posting is order-sensitive — sequential voucher numbers or first-seen-wins duplicate resolution — because nondeterministic completion order otherwise yields a non-reproducible audit trail.

Does chunking help if OCR is the bottleneck?

Yes, because chunking is what lets OCR run on every core at once instead of serially. A ProcessPoolExecutor gives each worker its own interpreter and GIL, so pool_width receipts are OCR’d simultaneously. Chunking bounds the memory that parallelism would otherwise consume; the two together turn a serial, memory-fragile job into a saturated, bounded one.

What is the difference between this and the queue-based approach?

A bounded queue decouples producers from consumers over time and across processes or hosts; chunking decides how a single large batch is sliced and how its results are stitched back together in order. They compose: the async batch queue delivers work, chunking sizes each unit for the pool, and backpressure on the worker pool keeps producers from outrunning consumers.

How do I stop one slow chunk from stalling everything?

A laggard cannot stall processing — other workers keep going — but it does hold the reorder buffer, since chunks after it cannot be emitted until it lands. Keep chunk_size small so a stall blocks fewer records, cap max_in_flight so the buffer cannot grow without bound, and route pathologically slow substrates to the faded-text restoration path rather than letting them dominate the tail.