Arbeiten mit großen JSON-Dateien: Optimierung und Best Practices

# Arbeiten mit großen JSON-Dateien

Große JSON-Dateien können eine Herausforderung darstellen. Dieser Leitfaden zeigt Ihnen, wie Sie auch mehrere Gigabyte große JSON-Dateien effizient verarbeiten können.

Das Problem mit großen JSON-Dateien

Typische Herausforderungen

• Speicherverbrauch - Gesamte Datei wird in den RAM geladen
• Parsing-Zeit - Langsames Parsen großer Strukturen
• Netzwerk-Overhead - Große Datenübertragungen
• Bearbeitungszeit - Langsame Operationen auf großen Objekten

Wann ist eine JSON-Datei "groß"?

• Klein: < 1 MB - Kein Problem
• Mittel: 1-100 MB - Optimierung sinnvoll
• Groß: 100 MB - 1 GB - Streaming empfohlen
• Sehr groß: > 1 GB - Spezialisierte Tools erforderlich

Streaming-Ansätze

JavaScript/Node.js Streaming

const fs = require('fs');
const JSONStream = require('JSONStream');

// Große Array-Datei streamen
const stream = fs.createReadStream('large-data.json', { encoding: 'utf8' });
const parser = JSONStream.parse('items.');


stream.pipe(parser);

parser.on('data', (item) => {
  // Jedes Item einzeln verarbeiten
  console.log(item);
  processItem(item);
});

parser.on('end', () => {
  console.log('Streaming abgeschlossen');
});

parser.on('error', (error) => {
  console.error('Streaming-Fehler:', error);
});

function processItem(item) {
  // Ihre Verarbeitung hier
  // Speicher wird für jedes Item freigegeben
}

import { createReadStream } from 'fs';
import { pipeline } from 'stream/promises';
import JSONStream from 'JSONStream';

async function processLargeJSON(filePath) {
  const readStream = createReadStream(filePath);
  const parseStream = JSONStream.parse('data.');

  try {
    await pipeline(
      readStream,
      parseStream,
      async function (source) {

        for await (const item of source) {
          // Async-Verarbeitung
          const processed = await processAsync(item);
          yield processed;
        }
      },
      async function (source) {
        for await (const item of source) {
          await saveToDatabase(item);
        }
      }
    );
    console.log('Verarbeitung abgeschlossen');
  } catch (error) {
    console.error('Pipeline-Fehler:', error);
  }
}

async function processAsync(item) {
  // Asynchrone Verarbeitung
  return { ...item, processed: true };
}

import ijson
from typing import Iterator

def stream_large_json(filepath: str) -> Iterator[dict]:
    """Streame große JSON-Datei Element für Element"""
    with open(filepath, 'rb') as file:
        # Parse items aus einem Array
        parser = ijson.items(file, 'items.item')

        for item in parser:
            yield item

# Verwendung
for item in stream_large_json('large-data.json'):
    print(f"Processing: {item['id']}")
    process_item(item)

def process_item(item):
    # Verarbeitung hier
    pass

import json
from typing import Iterator, Dict, Any

class JSONChunkProcessor:
    def __init__(self, filepath: str, chunk_size: int = 1000):
        self.filepath = filepath
        self.chunk_size = chunk_size

    def process_in_chunks(self) -> Iterator[list]:
        """Verarbeite JSON in Chunks"""
        chunk = []

        with open(self.filepath, 'r', encoding='utf-8') as f:
            # Annahme: JSON ist ein Array
            f.read(1)  # Öffnende Klammer überspringen

            decoder = json.JSONDecoder()
            buffer = ''

            for line in f:
                buffer += line

                while buffer:
                    buffer = buffer.lstrip()

                    if buffer.startswith(']'):
                        break

                    if buffer.startswith(','):
                        buffer = buffer[1:]
                        continue

                    try:
                        obj, idx = decoder.raw_decode(buffer)
                        chunk.append(obj)
                        buffer = buffer[idx:]

                        if len(chunk) >= self.chunk_size:
                            yield chunk
                            chunk = []
                    except json.JSONDecodeError:
                        # Brauchen mehr Daten
                        break

            if chunk:
                yield chunk

# Verwendung
processor = JSONChunkProcessor('large-array.json', chunk_size=100)

for chunk in processor.process_in_chunks():
    print(f"Verarbeite {len(chunk)} Items")
    # Chunk-Verarbeitung hier
    process_chunk(chunk)

{"id": 1, "name": "Alice", "value": 100}
{"id": 2, "name": "Bob", "value": 200}
{"id": 3, "name": "Charlie", "value": 300}

import json
from typing import Iterator, Callable, Dict, Any

class JSONLProcessor:
    @staticmethod
    def read(filepath: str) -> Iterator[Dict[str, Any]]:
        """Lese JSONL Zeile für Zeile"""
        with open(filepath, 'r', encoding='utf-8') as f:
            for line in f:
                if line.strip():
                    yield json.loads(line)

    @staticmethod
    def write(filepath: str, items: Iterator[Dict[str, Any]]):
        """Schreibe Items als JSONL"""
        with open(filepath, 'w', encoding='utf-8') as f:
            for item in items:
                f.write(json.dumps(item, ensure_ascii=False) + '\n')

    @staticmethod
    def transform(
        input_path: str,
        output_path: str,
        transform_fn: Callable[[Dict], Dict]
    ):
        """Transformiere JSONL im Streaming-Modus"""
        def transformed_items():
            for item in JSONLProcessor.read(input_path):
                yield transform_fn(item)

        JSONLProcessor.write(output_path, transformed_items())

    @staticmethod
    def filter(
        input_path: str,
        output_path: str,
        filter_fn: Callable[[Dict], bool]
    ):
        """Filtere JSONL im Streaming-Modus"""
        def filtered_items():
            for item in JSONLProcessor.read(input_path):
                if filter_fn(item):
                    yield item

        JSONLProcessor.write(output_path, filtered_items())

# Beispiele

# Transform
JSONLProcessor.transform(
    'input.jsonl',
    'output.jsonl',
    lambda item: {item, 'processed': True}

)

# Filter
JSONLProcessor.filter(
    'input.jsonl',
    'filtered.jsonl',
    lambda item: item.get('value', 0) > 100
)

# Lesen und verarbeiten
for record in JSONLProcessor.read('data.jsonl'):
    print(f"Processing {record['id']}")

import pandas as pd

# JSONL in chunks lesen
def read_jsonl_chunks(filepath: str, chunksize: int = 10000):
    """Lese JSONL in Chunks als DataFrames"""
    for chunk in pd.read_json(
        filepath, 
        lines=True, 
        chunksize=chunksize
    ):
        yield chunk

# Verwendung
for df_chunk in read_jsonl_chunks('large-data.jsonl', chunksize=5000):
    print(f"Processing chunk with {len(df_chunk)} rows")

    # Verarbeite DataFrame-Chunk
    result = df_chunk[df_chunk['value'] > 100]

    # Speichere Ergebnis
    result.to_json('output.jsonl', orient='records', lines=True, mode='a')

import orjson

# 2-3x schneller als standard json
data = {"key": "value", "numbers": list(range(10000))}

# Serialisieren
json_bytes = orjson.dumps(data)

# Deserialisieren
loaded = orjson.loads(json_bytes)

# Optionen
json_bytes = orjson.dumps(
    data,
    option=orjson.OPT_INDENT_2 | orjson.OPT_SORT_KEYS
)

import simdjson

# Noch schneller für große Dateien
parser = simdjson.Parser()

# Aus String
doc = parser.parse(json_string)

# Aus Datei (memory-mapped)
doc = parser.load('large.json')

# Zugriff
value = doc['key']['nested']['value']

import ujson

# Schnelleres encoding/decoding
data = {"large": "object"}

# Serialisieren
json_str = ujson.dumps(data)

# Deserialisieren
loaded = ujson.loads(json_str)

# Precision-Kontrolle für Floats
json_str = ujson.dumps(
    {"pi": 3.141592653589793},
    double_precision=2
)
# {"pi": 3.14}

import json
import gzip

# Schreiben mit Kompression
data = {"large": "dataset"  10000}

with gzip.open('data.json.gz', 'wt', encoding='utf-8') as f:
    json.dump(data, f)

# Lesen mit Kompression
with gzip.open('data.json.gz', 'rt', encoding='utf-8') as f:
    loaded = json.load(f)

# Streaming mit Kompression
import ijson

with gzip.open('large-data.json.gz', 'rb') as f:
    for item in ijson.items(f, 'items.item'):
        process(item)

bzip2 für bessere Kompression

import json
import bz2

# Noch bessere Kompression (aber langsamer)
with bz2.open('data.json.bz2', 'wt', encoding='utf-8') as f:
    json.dump(data, f)

with bz2.open('data.json.bz2', 'rt', encoding='utf-8') as f:
    loaded = json.load(f)

Memory-Mapped Files

JavaScript mit mmap

const mmap = require('mmap-io');
const fs = require('fs');

// Memory-mapped file für sehr große Dateien
const fd = fs.openSync('huge-file.json', 'r');
const stats = fs.fstatSync(fd);
const size = stats.size;

const buffer = mmap.map(size, mmap.PROT_READ, mmap.MAP_SHARED, fd, 0);

// Verarbeite Buffer in Chunks
const chunkSize = 1024  1024; // 1MB chunks

for (let offset = 0; offset < size; offset += chunkSize) {
  const chunk = buffer.slice(offset, Math.min(offset + chunkSize, size));
  processChunk(chunk);
}

mmap.advise(buffer, mmap.MADV_SEQUENTIAL);

import sqlite3
import json

def import_jsonl_to_sqlite(jsonl_path: str, db_path: str, table_name: str):
    """Importiere JSONL direkt in SQLite"""
    conn = sqlite3.connect(db_path)
    cursor = conn.cursor()

    # Tabelle erstellen
    cursor.execute(f'''
        CREATE TABLE IF NOT EXISTS {table_name} (
            id INTEGER PRIMARY KEY AUTOINCREMENT,
            data TEXT
        )
    ''')

    # Streaming-Import
    batch = []
    batch_size = 1000

    with open(jsonl_path, 'r', encoding='utf-8') as f:
        for line in f:
            if line.strip():
                batch.append((line.strip(),))

                if len(batch) >= batch_size:
                    cursor.executemany(
                        f'INSERT INTO {table_name} (data) VALUES (?)',
                        batch
                    )
                    conn.commit()
                    batch = []

        # Restliche Items
        if batch:
            cursor.executemany(
                f'INSERT INTO {table_name} (data) VALUES (?)',
                batch
            )
            conn.commit()

    conn.close()

# Verwendung
import_jsonl_to_sqlite('large-data.jsonl', 'data.db', 'records')

def query_json_from_sqlite(db_path: str, table_name: str, condition: str = None):
    """Query JSON aus SQLite mit JSON-Funktionen"""
    conn = sqlite3.connect(db_path)
    cursor = conn.cursor()

    query = f'''
        SELECT 
            json_extract(data, '$.id') as id,
            json_extract(data, '$.name') as name,
            data
        FROM {table_name}
    '''

    if condition:
        query += f' WHERE {condition}'

    cursor.execute(query)

    for row in cursor.fetchall():
        yield json.loads(row[2])

    conn.close()

# Verwendung
for record in query_json_from_sqlite(
    'data.db', 
    'records',
    "json_extract(data, '$.value') > 100"
):
    print(record)

import json
import orjson
import ujson
import time
from typing import Callable

def benchmark_parser(parser_fn: Callable, data: str, iterations: int = 1000):
    """Benchmark JSON-Parser"""
    start = time.time()

    for _ in range(iterations):
        parser_fn(data)

    elapsed = time.time() - start
    return elapsed / iterations

# Test-Daten
test_data = json.dumps({"key"  i: "value"  i for i in range(100)})


# Benchmarks
results = {
    'json': benchmark_parser(json.loads, test_data),
    'orjson': benchmark_parser(orjson.loads, test_data),
    'ujson': benchmark_parser(ujson.loads, test_data)
}

for parser, time_per_iter in results.items():
    print(f"{parser}: {time_per_iter1000:.3f}ms pro Iteration")

Memory Profiling

import tracemalloc
import json

def profile_memory(func):
    """Profile Speicherverbrauch"""
    tracemalloc.start()

    func()

    current, peak = tracemalloc.get_traced_memory()
    tracemalloc.stop()

    print(f"Current: {current / 1024 / 1024:.2f} MB")
    print(f"Peak: {peak / 1024 / 1024:.2f} MB")

# Test
def load_large_json():
    with open('large.json', 'r') as f:
        data = json.load(f)
    return data

profile_memory(load_large_json)

Best Practices

1. Richtige Tool-Auswahl

• Kleine Dateien (< 100MB): Standard JSON-Parser
• Mittlere Dateien (100MB-1GB): Streaming-Parser
• Große Dateien (> 1GB): JSONL + Datenbank oder spezialisierte Tools

2. Format-Optimierung

# Vermeiden: Verschachtelte Arrays in großen Dateien
{
  "data": [
    {"nested": [1, 2, 3, ...]},
    ...
  ]
}

# Besser: Flache Struktur mit JSONL
{"id": 1, "values": "1,2,3"}
{"id": 2, "values": "4,5,6"}

3. Lazy Loading

class LazyJSONLoader:
    def __init__(self, filepath: str):
        self.filepath = filepath
        self._data = None

    @property
    def data(self):
        if self._data is None:
            with open(self.filepath, 'r') as f:
                self._data = json.load(f)
        return self._data

    def clear_cache(self):
        self._data = None

# Verwendung
loader = LazyJSONLoader('large.json')
# Daten werden erst beim ersten Zugriff geladen
print(loader.data['key'])
# Cache freigeben
loader.clear_cache()

Zusammenfassung

Für große JSON-Dateien:

• Verwenden Sie Streaming für Dateien > 100MB
• Nutzen Sie JSONL für sehr große Datasets
• Optimierte Parser (orjson, simdjson) für bessere Performance
• Kompression spart Speicher und Bandbreite
• Datenbanken für querybare große Datasets
• Memory Profiling zur Identifikation von Problemen

Mit den richtigen Techniken können Sie auch Terabyte-große JSON-Datasets effizient verarbeiten.