Distributed Document Processing

"A PDF, waiting in a queue of ten million to become searchable. The scanner that made me was honest work; the OCR that reads me is somebody else's problem now."

A Scanned Invoice Between Parse and Chunk

The agentic application of this chapter is only as good as the corpus it can search. In Chapter 36 the corpus was the open web, crawled, deduplicated, and cleaned at petabyte scale. Here the corpus is an enterprise's own documents: contracts in PDF, policies in Word, financial models in Excel, decks in PowerPoint, and a long tail of scanned paper that arrived as images. The total volume is modest by web standards, perhaps a few million documents and a few terabytes, but the heterogeneity is brutal, and two enterprise constraints absent from a public crawl now dominate the design. First, access control: a chunk extracted from a document the requesting user may not read must never reach that user's agent, so permissions are not an afterthought but a field that flows through the entire pipeline. Second, structure: enterprise documents carry tables, forms, and headings whose meaning is lost if the layout is flattened to a wall of text. This section turns that pile into a clean, permissioned, retrievable chunk store, and it does so as a distributed batch job because the per-document work is independent and embarrassingly parallel.

1. Parsing a Corpus of Mixed Formats Beginner

The first stage takes a file and returns text plus structure. A born-digital PDF carries a text layer that can be read directly, but its reading order is rarely linear: multi-column layouts, headers, footers, and footnotes interleave, and a naive extraction yields scrambled prose. Layout-aware extraction recovers the intended order by reasoning about the geometry of text blocks on the page, keeping a table's cells together and separating a sidebar from the body. Office formats are friendlier because the structure is explicit in the file: a DOCX exposes headings and paragraphs, an XLSX exposes sheets and cells, a PPTX exposes slides and their text frames. The hard case is the scan, a PDF or image with no text layer at all, where optical character recognition (OCR) must convert pixels to characters before anything else can proceed. OCR is the most compute-heavy step in the pipeline and the one with the most variable cost per page, which matters for the sizing argument of Section 4.

Tables deserve their own treatment because they are where enterprise meaning concentrates and where flattening does the most damage. A pricing table read row-major into a single paragraph loses the column that says whether a number is a unit price or a total. Layout-aware parsers emit tables as structured objects (rows and cells, or serialized to a small markup such as HTML or Markdown) so that the retriever later sees "Region: EMEA, Q3 revenue: 4.2M" rather than an undifferentiated string of digits. This is a point of contact with the cleaning discipline of Section 36.3, where boilerplate stripping and normalization were developed for web pages; here the same cleaning runs, but the input is a parser's structured output rather than raw HTML, and the boilerplate to strip is headers, footers, watermarks, and confidentiality stamps rather than navigation bars and ad slots.

2. Chunking Strategies and Why They Shape Retrieval Intermediate

A parsed document is usually too long to embed or retrieve as a single unit, so it is split into chunks, the atomic pieces the retriever indexes and the agent reads. The simplest strategy is fixed-size chunking: cut the text into windows of a fixed length with a fixed overlap between neighbors. The overlap exists so that a sentence straddling a boundary survives intact in at least one chunk, at the cost of storing some text twice. If each document yields on average $\bar{c}$ chunks, the corpus produces a total chunk count of

for $D$ documents, and this number, often in the tens of millions, sets the size of the vector index that Section 40.3 must build and Chapter 25 must serve. With a chunk window of $s$ characters and an overlap of $o$, a document of length $L$ produces about $\lceil L / (s - o) \rceil$ chunks, and the fraction of stored text that is duplicated rises with the overlap ratio $o / s$. Larger chunks pack more context per retrieval but blur the granularity, so a query about one clause drags in a whole page of unrelated text; smaller chunks sharpen granularity but risk severing the context a clause needs to be understood. This is the central chunk-size trade-off, and there is no universally right answer, only a right answer for a given retriever and query distribution.

Semantic chunking respects natural boundaries instead of a character count: it packs whole sentences (or paragraphs, or sections) until a size budget is reached, so no chunk ever ends mid-sentence. Hierarchical, or parent-child, chunking goes further by storing two granularities at once: small child chunks are indexed for precise retrieval, but when one is matched the retriever returns its larger parent (the surrounding section) so the agent reads with full context. The slogan is that you retrieve on the small piece and read on the big one. Which strategy wins is an empirical question tied to the corpus and the agent, but the structural point is universal: chunk boundaries are the unit of retrieval, so a boundary in the wrong place is a fact the agent can never assemble. The demonstration below makes the fixed-versus-semantic contrast concrete on a short policy document.

import re

# A small enterprise document body (a policy excerpt) used to compare chunkers.
text = (
    "Employees may request remote-work arrangements through the HR portal. "
    "Requests must be approved by a direct manager and recorded in the system of record. "
    "Reimbursement for home-office equipment is capped at eight hundred dollars per fiscal year. "
    "Receipts are required for any single item above one hundred dollars. "
    "Access to confidential customer data from outside the corporate network requires a hardware token. "
    "Violations of the data-handling policy are escalated to the security team within one business day. "
    "All policy exceptions must be documented and reviewed quarterly by the compliance office. "
    "This document supersedes the prior remote-work memorandum dated the previous fiscal year."
)

def fixed_chunks(s, size, overlap):
    """Fixed-size character windows with overlap; the standard RAG default."""
    step = size - overlap
    out, i = [], 0
    while i < len(s):
        out.append(s[i:i + size])
        i += step
    return out

def sentence_chunks(s, max_chars):
    """Semantic-ish split: pack whole sentences until the budget is hit."""
    sents = re.findall(r"[^.]*\.", s)
    out, cur = [], ""
    for sent in sents:
        sent = sent.strip()
        if cur and len(cur) + 1 + len(sent) > max_chars:
            out.append(cur); cur = sent
        else:
            cur = (cur + " " + sent).strip()
    if cur:
        out.append(cur)
    return out

SIZE, OVERLAP = 200, 40
fx = fixed_chunks(text, SIZE, OVERLAP)
sx = sentence_chunks(text, SIZE)
print(f"document length (chars) : {len(text)}")
print(f"fixed  size={SIZE} overlap={OVERLAP}")
print(f"  chunks                : {len(fx)}")
print(f"  sizes                 : {[len(c) for c in fx]}")
dup = sum(min(OVERLAP, len(fx[i])) for i in range(1, len(fx)))   # repeated chars
print(f"  duplicated chars      : {dup}  ({dup / len(text):.0%} redundancy)")
print(f"sentence-packed max={SIZE}")
print(f"  chunks                : {len(sx)}")
print(f"  sizes                 : {[len(c) for c in sx]}")
print(f"  mid-sentence cuts     : 0 (boundaries respected)")

# Corpus sizing: chunk count C = D * avg_chunks_per_doc.
DOCS, avg_chunks = 2_000_000, len(fx)
total = DOCS * avg_chunks
WORKERS, DOCS_PER_SEC = 64, 12.0
throughput = WORKERS * DOCS_PER_SEC          # docs/s = workers * per-worker rate
print(f"corpus: {DOCS:,} docs x {avg_chunks} chunks = {total:,} chunks")
print(f"throughput: {WORKERS} workers x {DOCS_PER_SEC} docs/s = {throughput:,.0f} docs/s")
print(f"wall-clock to ingest    : {DOCS / throughput / 3600:.2f} h")

document length (chars) : 692
fixed  size=200 overlap=40
  chunks                : 5
  sizes                 : [200, 200, 200, 200, 52]
  duplicated chars      : 160  (23% redundancy)
sentence-packed max=200
  chunks                : 4
  sizes                 : [153, 160, 197, 179]
  mid-sentence cuts     : 0 (boundaries respected)
corpus: 2,000,000 docs x 5 chunks = 10,000,000 chunks
throughput: 64 workers x 12.0 docs/s = 768 docs/s
wall-clock to ingest    : 0.72 h

The numbers in Output 40.2.1 make the trade-off legible. The fixed chunker's overlap buys boundary robustness at a twenty-three percent storage tax, money the system pays on every one of the ten million chunks and again in the vector index. The sentence-packed variant spends no overlap and respects boundaries, but its chunks vary in size and it depends on reliable sentence segmentation, which OCR noise can break. Choosing between them is the kind of empirical decision that retrieval evaluation (the metrics of Chapter 5) is meant to settle, not a matter of taste.

from unstructured.partition.auto import partition
from langchain_text_splitters import RecursiveCharacterTextSplitter

elements = partition(filename="policy.pdf")                 # parse + OCR if needed
text = "\n\n".join(e.text for e in elements if e.text)

splitter = RecursiveCharacterTextSplitter(
    chunk_size=800, chunk_overlap=120,                       # tries paragraph, then
    separators=["\n\n", "\n", ". ", " "])                    # sentence, then word breaks
chunks = splitter.split_text(text)                           # boundary-aware, no manual loop

3. Metadata, Permissions, and Structured Tables Intermediate

A chunk is not just text. Every chunk carries a metadata record that the retriever filters on and the agent reasons with: the source document identifier and version, the page or section it came from, the document type, timestamps, and, in the enterprise, the access-control list inherited from the source file. This last field is what separates an enterprise pipeline from the public crawl of Section 36.2, where everything was world-readable. Here a chunk drawn from a board-only memorandum must carry the memorandum's permissions, so that retrieval can be filtered by the requesting user's identity before any text is returned. Access control enforced at query time on a chunk that lacks its permission tag is no access control at all; the tag must be attached at ingestion, when the source file's permissions are still in hand, and it must survive every later stage. The retriever of Chapter 25 then treats the permission field as a hard pre-filter on the candidate set.

Structured tables also live in metadata, or alongside the text in a form the agent can parse. A chunk that contains a financial table is most useful when the table's structure is preserved and a short natural-language summary sits beside it, so that the retriever can match on the summary's words while the agent reads the exact figures from the structure. Metadata is likewise what makes incremental ingestion possible: the document version and a content hash stored on each chunk let the next run detect what changed. The thread to keep in view is that metadata and chunk boundaries jointly determine the reachable set: the retriever can only surface a fact that lives inside some chunk whose metadata passes the query's filters, so a missing permission tag hides a chunk the user was entitled to, and a wrong document type filters it out of every relevant query.

4. The Distributed Batch Pipeline and Its Sizing Intermediate

Because each document is processed independently, the pipeline is a textbook map job. Expressed in the Spark idiom of Chapter 7, the corpus is a distributed collection of document paths, a single transformation maps each path through parse, clean, chunk, and tag, the result is flattened (one document yields many chunks, so the map is a flatMap), and the chunks are written to the store. No stage requires data from another document, so there is no shuffle and the job's throughput is, to first order, linear in the worker count. With $W$ workers each processing $r$ documents per second, aggregate throughput is

so the wall-clock to ingest the whole corpus falls inversely with the number of workers until some shared resource (the document store's read bandwidth from Chapter 8, or the chunk store's write bandwidth) becomes the bottleneck. Output 40.2.1 instantiates these formulas: at $W = 64$ workers and $r = 12$ documents per second the pipeline clears two million documents in about forty-three minutes. The per-document rate $r$ is dominated by the slowest stage, which for a scanned-heavy corpus is OCR, so the cluster is provisioned around the OCR cost and the cheaper born-digital documents ride along for free. This is the same batch-sizing logic that Chapter 3 made rigorous, applied to a document corpus.

5. Incremental Ingestion for a Living Corpus Advanced

An enterprise corpus is never frozen. Contracts are amended, policies are revised, new decks land every day, and a few documents are deleted or have their permissions changed. Re-ingesting the entire corpus on every change would be wasteful and would churn the vector index needlessly, so ingestion becomes incremental in the spirit of Chapter 9: a periodic or event-driven job processes only the documents whose content hash or version changed since the last run, mapping each through the same parse-clean-chunk-tag path and then upserting its chunks, replacing the old chunks for that document and removing any the new version no longer produces. The content hash carried in each chunk's metadata (Section 3) is what makes the change set cheap to compute. Deletions and permission changes are the subtle cases: a deleted document must have all its chunks purged from the store and the index, and a permission downgrade must propagate to every chunk so the retriever's pre-filter immediately reflects the new access rules. Incremental ingestion is therefore not a smaller version of the batch job but the batch map restricted to a change set, with an upsert and a tombstone path bolted on, and it shares all the parsing and chunking logic verbatim.

With a clean, permissioned, incrementally maintained chunk store in place, the corpus is ready to be made searchable. The next step is to turn each chunk into a vector the retriever can match against, the distributed embedding stage that Section 40.3 builds, where the ten million chunks counted in Output 40.2.1 become ten million vectors in the index that Chapter 25 serves to the agent.