How to Process Multi-Gigabyte JSON Files in Python: Memory-Efficient Techniques That Work

import ijson

# Process a large JSON file containing an array of objects
with open('massive_dataset.json', 'rb') as f:
    # Extract only objects within the "customers" array
    for customer in ijson.items(f, 'customers.item'):
        # Process each customer individually
        name = customer.get('name')
        email = customer.get('email')
        process_customer_data(name, email)

import json

def process_json_lines(filename):
    with open(filename, 'r') as f:
        for line in f:
            if line.strip():  # Skip empty lines
                record = json.loads(line)
                yield record

# Usage
for item in process_json_lines('large_records.jsonl'):
    # Process each item with minimal memory overhead
    print(item['id'])

import mmap
import json
import re

def find_json_objects(filename, pattern):
    with open(filename, 'r+b') as f:
        # Create memory-mapped file
        mm = mmap.mmap(f.fileno(), 0)

        # Search for pattern in the file
        pattern_compiled = re.compile(pattern.encode())

        # Find all matches
        for match in pattern_compiled.finditer(mm):
            # Extract the JSON object containing the match
            start_pos = mm.rfind(b'{', 0, match.start())
            end_pos = mm.find(b'}', match.end())

            if start_pos != -1 and end_pos != -1:
                json_bytes = mm[start_pos:end_pos+1]
                try:
                    yield json.loads(json_bytes)
                except json.JSONDecodeError:
                    # Handle parsing errors
                    pass

        mm.close()

# Usage
for obj in find_json_objects('analytics_data.json', 'error_code'):
    log_error(obj)

import json

def process_in_chunks(filename, chunk_size=1000):
    chunk = []
    with open(filename, 'r') as f:
        # Assuming JSON file contains an array of objects
        f.readline()  # Skip the opening '['

        for line in f:
            line = line.strip()
            if line.endswith(','):
                line = line[:-1]

            if line and line != ']':
                try:
                    item = json.loads(line)
                    chunk.append(item)

                    if len(chunk) >= chunk_size:
                        yield chunk
                        chunk = []
                except json.JSONDecodeError:
                    # Handle malformed JSON
                    continue

        if chunk:  # Don't forget the last chunk
            yield chunk

# Usage
for batch in process_in_chunks('product_catalog.json', 500):
    db.bulk_insert(batch)

import json
import gzip

def process_compressed_json(filename):
    with gzip.open(filename, 'rt', encoding='utf-8') as f:
        # For a JSON array structure
        data = json.load(f)

        for item in data:
            yield item

# Alternatively, for line-delimited JSON
def process_compressed_jsonl(filename):
    with gzip.open(filename, 'rt', encoding='utf-8') as f:
        for line in f:
            if line.strip():
                yield json.loads(line)

# Usage
for record in process_compressed_jsonl('logs.jsonl.gz'):
    analyze_log_entry(record)

import json
from jsonpath_ng import parse

def extract_with_jsonpath(filename, json_path):
    with open(filename, 'r') as f:
        data = json.load(f)

    # Compile the JSONPath expression
    jsonpath_expr = parse(json_path)

    # Find all matches
    return [match.value for match in jsonpath_expr.find(data)]

# Usage - extract all prices from a product catalog
prices = extract_with_jsonpath('catalog.json', '$..price')

def extract_with_jsonpath_chunked(filename, json_path, chunk_size=100):
    jsonpath_expr = parse(json_path)

    for chunk in process_in_chunks(filename, chunk_size):
        for item in chunk:
            for match in jsonpath_expr.find(item):
                yield match.value

import json
from concurrent.futures import ProcessPoolExecutor
import os

def process_partition(filename, start_pos, end_pos):
    results = []
    with open(filename, 'r') as f:
        f.seek(start_pos)

        # Read to the first complete line if not at file start
        if start_pos != 0:
            f.readline()

        line = f.readline()
        while line and f.tell() <= end_pos:
            try:
                record = json.loads(line)
                # Process the record
                results.append(transform_record(record))
            except json.JSONDecodeError:
                pass
            line = f.readline()

    return results

def parallel_process_json(filename, num_workers=None):
    if num_workers is None:
        num_workers = os.cpu_count()

    # Get file size
    file_size = os.path.getsize(filename)

    # Calculate partition sizes
    chunk_size = file_size // num_workers

    # Create tasks
    tasks = []
    for i in range(num_workers):
        start = i * chunk_size
        end = (i + 1) * chunk_size if i < num_workers - 1 else file_size
        tasks.append((filename, start, end))

    # Process in parallel
    all_results = []
    with ProcessPoolExecutor(max_workers=num_workers) as executor:
        for result in executor.map(lambda p: process_partition(*p), tasks):
            all_results.extend(result)

    return all_results

import ijson
import gzip
from concurrent.futures import ThreadPoolExecutor

def process_compressed_stream(filename, batch_size=1000):
    batch = []

    with gzip.open(filename, 'rb') as f:
        # Stream-parse the JSON data
        parser = ijson.items(f, 'item')

        for item in parser:
            batch.append(item)

            if len(batch) >= batch_size:
                yield batch
                batch = []

    if batch:  # Don't forget the last batch
        yield batch

def process_batch(batch):
    # Process a batch of records
    results = []
    for item in batch:
        # Do some transformation
        transformed = transform_data(item)
        results.append(transformed)

    # Bulk save to database
    save_to_database(results)
    return len(results)

def main():
    filename = 'massive_dataset.json.gz'
    total_processed = 0

    # Create a thread pool
    with ThreadPoolExecutor(max_workers=4) as executor:
        futures = []

        # Submit batch processing tasks
        for batch in process_compressed_stream(filename):
            future = executor.submit(process_batch, batch)
            futures.append(future)

        # Collect results
        for future in futures:
            total_processed += future.result()

    print(f"Processed {total_processed} records")

if __name__ == "__main__":
    main()

def transform_stream(data_stream):
    for item in data_stream:
        # Apply transformations
        if 'name' in item:
            item['name'] = item['name'].upper()

        if 'timestamp' in item:
            item['date'] = convert_timestamp_to_date(item['timestamp'])

        yield item

# Usage with our previous function
for batch in transform_stream(process_json_lines('data.jsonl')):
    write_to_output(batch)

def analyze_user_interactions():
    # Process multiple large files
    file_list = glob.glob('user_data_*.json.gz')

    total_interactions = 0
    user_stats = {}

    for filename in file_list:
        print(f"Processing {filename}")

        # Process file with our stream processor
        for batch in process_compressed_stream(filename):
            # Update statistics
            for interaction in batch:
                user_id = interaction.get('user_id')
                action = interaction.get('action')

                if user_id and action:
                    if user_id not in user_stats:
                        user_stats[user_id] = {'actions': {}}

                    if action not in user_stats[user_id]['actions']:
                        user_stats[user_id]['actions'][action] = 0

                    user_stats[user_id]['actions'][action] += 1
                    total_interactions += 1

    print(f"Analyzed {total_interactions} interactions across {len(user_stats)} users")
    return user_stats