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