Fast NDJSON Processor

A high-performance Python library for processing NDJSON (Newline Delimited JSON) files with multiprocessing support and streaming capabilities.

🚀 Key Features

• ⚡ High Performance: Multiprocessing with chunking for optimal speed
• 🔄 Multiple Processing Modes: Parallel, Sequential, and Streaming
• 📦 Chunk Processing: Process batches of records for aggregations and transformations
• 📝 Type Safety: Full type annotations and mypy support

📦 Installation

pip install fast-ndjson-processor

🎯 Quick Start

Basic Processing

from fast_ndjson_processor import FastNDJSONProcessor

# Create processor
processor = FastNDJSONProcessor(n_workers=4, show_progress=True)

# Define your processing function
def extract_user_data(record):
    return {
        "user_id": record["id"],
        "name": record["name"],
        "email": record.get("email")
    }

# Process the file
results = processor.process_file("users.ndjson", extract_user_data)
print(f"Processed {len(results)} records")

Streaming for Large Files

# Memory-efficient streaming
for record in processor.stream_file("large_file.ndjson"):
    # Process one record at a time
    process_record(record)

Chunk Processing

# Process chunks of records for aggregations
def analyze_chunk(records):
    """Process a batch of records at once"""
    return {
        "count": len(records),
        "total_value": sum(r.get("value", 0) for r in records),
        "categories": list(set(r.get("category") for r in records))
    }

# Use the class method
results = processor.process_file_chunks("data.ndjson", analyze_chunk)

Writing NDJSON Files

from fast_ndjson_processor import FastNDJSONWriter

# Write data to NDJSON file
data = [
    {"id": 1, "name": "Alice"},
    {"id": 2, "name": "Bob"}
]

with FastNDJSONWriter("output.ndjson") as writer:
    writer.save(data)

📖 Documentation

Processing Modes

1. Parallel Mode (Default)

from fast_ndjson_processor import ProcessorMode

results = processor.process_file(
    "data.ndjson", 
    handler_function, 
    mode=ProcessorMode.PARALLEL
)

2. Sequential Mode

results = processor.process_file(
    "data.ndjson", 
    handler_function, 
    mode=ProcessorMode.SEQUENTIAL
)

3. Streaming Mode

results = processor.process_file(
    "data.ndjson", 
    handler_function, 
    mode=ProcessorMode.STREAMING
)

🧪 Testing

# Run all tests
pytest

# Run with coverage
pytest --cov=fast_ndjson_processor --cov-report=html

# Run specific test categories
pytest -m "not slow"  # Skip slow tests
pytest -m "integration"  # Run only integration tests

🤝 Contributing

Please see Contributing Guide for details.

Development Setup

# Clone the repository
git clone https://github.com/yourusername/fast_ndjson_processor.git
cd fast_ndjson_processor

# Create virtual environment
python -m venv venv
source venv/bin/activate  # On Windows: venv\Scripts\activate

# Install in development mode
pip install -e .[dev]

# Run pre-commit hooks
pre-commit install

# Run tests
pytest

Code Quality

# Format code
black .
isort .

# Lint code
flake8 fast_ndjson_processor tests

# Type checking
mypy fast_ndjson_processor

# Security check
bandit -r fast_ndjson_processor/

📄 License

This project is licensed under GPLv3 - see the LICENSE file for details.

🙏 Acknowledgments

• Inspired by pushshift/ndjson_processor and pushshift/Parallel-NDJSON-Reader.
• Built with ijl/orjson for fast JSON processing. Thanks umarbutler/orjsonl for providing the reference for orjson usage in NDJSON processing.