Production Patterns

Moving from experimentation to production requires attention to performance, error handling, and data safety. This lesson covers patterns for reliable PDF processing at scale.

Batch Processing

The JVM Overhead Problem

Each convert() call spawns a new Java process:

# SLOW: 3 files × 2s JVM startup = 6s overhead
for pdf in ['a.pdf', 'b.pdf', 'c.pdf']:
    opendataloader_pdf.convert(input_path=pdf, output_dir="output/")

Batching Files

Pass multiple files or a directory to a single call:

import opendataloader_pdf

# FAST: Single JVM for all files
opendataloader_pdf.convert(
    input_path="pdfs/",  # Directory of PDFs
    output_dir="output/",
    format="json"
)

# Or explicit file list
opendataloader_pdf.convert(
    input_path=["a.pdf", "b.pdf", "c.pdf"],
    output_dir="output/",
    format="json"
)

Optimal Batch Sizes

import os
from pathlib import Path

def batch_process(pdf_dir, output_dir, batch_size=50):
    """Process PDFs in optimal batches."""
    pdf_files = list(Path(pdf_dir).glob("*.pdf"))
    total = len(pdf_files)

    for i in range(0, total, batch_size):
        batch = pdf_files[i:i + batch_size]
        print(f"Processing batch {i//batch_size + 1}: {len(batch)} files")

        opendataloader_pdf.convert(
            input_path=[str(p) for p in batch],
            output_dir=output_dir,
            format="json"
        )

    print(f"Completed: {total} files")

# Process thousands of PDFs efficiently
batch_process("documents/", "output/", batch_size=50)

Error Handling

Catching Conversion Errors

import opendataloader_pdf

def safe_convert(pdf_path, output_dir):
    """Convert with error handling."""
    try:
        opendataloader_pdf.convert(
            input_path=pdf_path,
            output_dir=output_dir,
            format="json"
        )
        return {"status": "success", "file": pdf_path}

    except Exception as e:
        return {
            "status": "error",
            "file": pdf_path,
            "error": str(e)
        }

# Process with error tracking
results = []
for pdf in pdf_files:
    result = safe_convert(pdf, "output/")
    results.append(result)

# Report errors
errors = [r for r in results if r['status'] == 'error']
print(f"Failed: {len(errors)}/{len(results)}")
for e in errors:
    print(f"  {e['file']}: {e['error']}")

Fallback Strategies

def convert_with_fallback(pdf_path, output_dir):
    """Try multiple strategies for robust conversion."""

    # Strategy 1: Normal conversion
    try:
        opendataloader_pdf.convert(
            input_path=pdf_path,
            output_dir=output_dir,
            format="json"
        )
        return "success"
    except Exception as e:
        print(f"Normal conversion failed: {e}")

    # Strategy 2: Force OCR
    try:
        opendataloader_pdf.convert(
            input_path=pdf_path,
            output_dir=output_dir,
            format="json",
            force_ocr=True
        )
        return "success_ocr"
    except Exception as e:
        print(f"OCR conversion failed: {e}")

    # Strategy 3: Hybrid mode
    try:
        opendataloader_pdf.convert(
            input_path=pdf_path,
            output_dir=output_dir,
            format="json",
            hybrid=True
        )
        return "success_hybrid"
    except Exception as e:
        print(f"Hybrid conversion failed: {e}")

    return "failed"

PII Sanitization

Automatic PII Redaction

opendataloader_pdf.convert(
    input_path="contracts/",
    output_dir="output/",
    format="json",
    sanitize=True  # Redact PII
)

What Gets Sanitized

Custom Sanitization

For domain-specific PII:

import re
import json

def custom_sanitize(json_path):
    """Add custom PII sanitization."""
    with open(json_path, 'r') as f:
        data = json.load(f)

    # Custom patterns
    patterns = {
        'EMPLOYEE_ID': r'EMP-\d{6}',
        'PROJECT_CODE': r'PRJ-[A-Z]{3}-\d{4}',
        'INTERNAL_IP': r'10\.\d+\.\d+\.\d+'
    }

    def sanitize_text(text):
        for name, pattern in patterns.items():
            text = re.sub(pattern, f'[{name}]', text)
        return text

    # Apply to all elements
    for element in data.get('kids', []):
        if 'content' in element:
            element['content'] = sanitize_text(element['content'])

    # Save sanitized version
    with open(json_path.replace('.json', '_sanitized.json'), 'w') as f:
        json.dump(data, f, indent=2)

# Apply after conversion
custom_sanitize('output/document.json')

Performance Monitoring

Timing Operations

import time
from pathlib import Path

def timed_convert(pdf_path, output_dir):
    """Convert and measure performance."""
    start = time.time()

    opendataloader_pdf.convert(
        input_path=pdf_path,
        output_dir=output_dir,
        format="json"
    )

    elapsed = time.time() - start

    # Get file stats
    pdf_size = Path(pdf_path).stat().st_size / 1024  # KB

    return {
        "file": pdf_path,
        "size_kb": pdf_size,
        "time_seconds": elapsed,
        "kb_per_second": pdf_size / elapsed
    }

# Benchmark your pipeline
stats = [timed_convert(pdf, "output/") for pdf in pdf_files]

# Calculate averages
avg_time = sum(s['time_seconds'] for s in stats) / len(stats)
avg_throughput = sum(s['kb_per_second'] for s in stats) / len(stats)

print(f"Average time: {avg_time:.2f}s")
print(f"Average throughput: {avg_throughput:.0f} KB/s")

Memory Management

For very large batches, process in chunks:

import gc

def memory_safe_batch(pdf_dir, output_dir, chunk_size=25):
    """Process with memory cleanup between chunks."""
    pdf_files = list(Path(pdf_dir).glob("*.pdf"))

    for i in range(0, len(pdf_files), chunk_size):
        chunk = pdf_files[i:i + chunk_size]
        print(f"Processing chunk {i//chunk_size + 1}")

        opendataloader_pdf.convert(
            input_path=[str(p) for p in chunk],
            output_dir=output_dir,
            format="json"
        )

        # Force garbage collection between chunks
        gc.collect()

        print(f"  Completed {min(i + chunk_size, len(pdf_files))}/{len(pdf_files)}")

Production Pipeline Example

Putting it all together:

import opendataloader_pdf
import json
from pathlib import Path
from datetime import datetime
import logging

# Setup logging
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

class PDFPipeline:
    def __init__(self, input_dir, output_dir, batch_size=50):
        self.input_dir = Path(input_dir)
        self.output_dir = Path(output_dir)
        self.batch_size = batch_size
        self.results = []

    def run(self, sanitize=False, hybrid=False):
        """Run the full pipeline."""
        pdf_files = list(self.input_dir.glob("*.pdf"))
        logger.info(f"Found {len(pdf_files)} PDFs to process")

        # Process in batches
        for i in range(0, len(pdf_files), self.batch_size):
            batch = pdf_files[i:i + self.batch_size]
            self._process_batch(batch, sanitize, hybrid)

        # Generate report
        self._generate_report()

        return self.results

    def _process_batch(self, batch, sanitize, hybrid):
        """Process a single batch."""
        batch_num = len(self.results) // self.batch_size + 1
        logger.info(f"Processing batch {batch_num}: {len(batch)} files")

        try:
            opendataloader_pdf.convert(
                input_path=[str(p) for p in batch],
                output_dir=str(self.output_dir),
                format="json",
                sanitize=sanitize,
                hybrid=hybrid
            )

            for pdf in batch:
                self.results.append({
                    "file": str(pdf),
                    "status": "success",
                    "timestamp": datetime.now().isoformat()
                })

        except Exception as e:
            logger.error(f"Batch failed: {e}")
            # Retry individually
            for pdf in batch:
                self._process_single(pdf, sanitize, hybrid)

    def _process_single(self, pdf_path, sanitize, hybrid):
        """Process a single file with fallback."""
        try:
            opendataloader_pdf.convert(
                input_path=str(pdf_path),
                output_dir=str(self.output_dir),
                format="json",
                sanitize=sanitize,
                hybrid=hybrid
            )
            status = "success"
        except Exception as e:
            status = f"error: {e}"

        self.results.append({
            "file": str(pdf_path),
            "status": status,
            "timestamp": datetime.now().isoformat()
        })

    def _generate_report(self):
        """Generate processing report."""
        success = sum(1 for r in self.results if r['status'] == 'success')
        failed = len(self.results) - success

        report = {
            "total": len(self.results),
            "success": success,
            "failed": failed,
            "success_rate": f"{100 * success / len(self.results):.1f}%",
            "details": self.results
        }

        report_path = self.output_dir / "pipeline_report.json"
        with open(report_path, 'w') as f:
            json.dump(report, f, indent=2)

        logger.info(f"Pipeline complete: {success}/{len(self.results)} successful")
        logger.info(f"Report saved to: {report_path}")

# Run production pipeline
pipeline = PDFPipeline(
    input_dir="incoming_documents/",
    output_dir="processed/",
    batch_size=50
)

results = pipeline.run(sanitize=True, hybrid=False)

Challenge: Build a Resilient Pipeline

Create a production-ready pipeline that:

• Processes a directory of PDFs with progress tracking
• Implements retry logic with exponential backoff
• Generates a detailed report with success/failure stats
• Handles memory efficiently for large batches

# Your implementation here
# Test with: python pipeline.py ./test_pdfs/ ./output/

Expected output:

PDF Pipeline Started
===================
Input: ./test_pdfs/ (127 files)
Output: ./output/
Batch size: 50

Batch 1/3: Processing 50 files...
  ✓ 50/50 successful (2.3s)

Batch 2/3: Processing 50 files...
  ✓ 48/50 successful (2.1s)
  ⚠ 2 files failed, retrying...
  ✓ 1/2 recovered with OCR
  ✗ 1/2 failed permanently

Batch 3/3: Processing 27 files...
  ✓ 27/27 successful (1.2s)

Pipeline Complete
================
Total: 127 files
Success: 126 (99.2%)
Failed: 1 (0.8%)

Report: ./output/pipeline_report.json

Congratulations!

You've completed the OpenDataLoader PDF course. You now know how to:

• Extract structured data from any PDF
• Handle tables, formulas, and images
• Process scanned documents with OCR
• Build production-ready pipelines

For more resources:

• [OpenDataLoader GitHub](https://github.com/opendataloader-project/opendataloader-pdf)
• [API Documentation](https://opendataloader.dev/docs)
• [Benchmark Results](https://opendataloader.dev/benchmarks)