Формат JSON Lines: работа с потоковыми JSON данными в Python

Расширенные примеры работы с JSON Lines

Ниже приведены примеры для нестандартных сценариев использования.

1. Объединение нескольких JSON Lines файлов в один

import json

def merge_jsonl_files(input_files, output_file):
    with open(output_file, 'w', encoding='utf-8') as out:
        for fname in input_files:
            with open(fname, 'r', encoding='utf-8') as inf:
                for line in inf:
                    if line.strip():
                        out.write(line)

merge_jsonl_files(['part1.jsonl', 'part2.jsonl'], 'merged.jsonl')
print("Файлы объединены")

(результат: создан merged.jsonl, содержащий все строки)

2. Параллельная обработка с multiprocessing

import json
import multiprocessing as mp

def process_line(line):
    obj = json.loads(line.strip())
    obj['triple'] = obj['value'] * 3
    return json.dumps(obj, ensure_ascii=False) + '\n'

def parallel_process_jsonl(input_file, output_file, num_workers=4):
    with open(input_file, 'r') as f:
        lines = f.readlines()
    with mp.Pool(num_workers) as pool:
        processed = pool.map(process_line, lines)
    with open(output_file, 'w') as f:
        f.writelines(processed)

parallel_process_jsonl('data.jsonl', 'processed.jsonl')
print("Параллельная обработка завершена")

3. Асинхронное чтение с asyncio

import asyncio
import json

def read_line_gen(filepath):
    with open(filepath, 'r') as f:
        for line in f:
            if line.strip():
                yield line

async def process_async(filepath):
    loop = asyncio.get_event_loop()
    with open(filepath, 'r') as f:
        async for line in async_iter_lines(f):
            obj = json.loads(line.strip())
            # асинхронная обработка
            print(obj)

# Вспомогательная функция
def async_iter_lines(file_obj):
    async def wrapper():
        for line in file_obj:
            yield line
    return wrapper()

asyncio.run(process_async('data.jsonl'))

(Асинхронный вывод может быть перемешан, но каждый объект обработан)

4. Валидация схемы записей при чтении

import json
from jsonschema import validate, ValidationError

SCHEMA = {
    "type": "object",
    "properties": {
        "id": {"type": "integer"},
        "name": {"type": "string"}
    },
    "required": ["id", "name"]
}

def read_validated(filepath):
    with open(filepath, 'r') as f:
        for line in f:
            line = line.strip()
            if not line:
                continue
            obj = json.loads(line)
            try:
                validate(instance=obj, schema=SCHEMA)
                yield obj
            except ValidationError as e:
                print(f"Невалидная запись: {e}")

for rec in read_validated('data.jsonl'):
    print(rec['name'])

5. Конвертация CSV в JSON Lines

import csv
import json

def csv_to_jsonl(csv_file, jsonl_file, delimiter=','):
    with open(csv_file, 'r', newline='', encoding='utf-8') as csvf:
        reader = csv.DictReader(csvf, delimiter=delimiter)
        with open(jsonl_file, 'w', encoding='utf-8') as jsonlf:
            for row in reader:
                jsonlf.write(json.dumps(row, ensure_ascii=False) + '\n')

csv_to_jsonl('input.csv', 'output.jsonl')
print("Конвертация завершена")

(создан файл output.jsonl, каждая строка - JSON объект с ключами из заголовков CSV)

6. Использование памяти через mmap для больших файлов

import mmap
import json

def fast_read_jsonl(filepath):
    with open(filepath, 'r+b') as f:
        mm = mmap.mmap(f.fileno(), 0, access=mmap.ACCESS_READ)
        pos = 0
        while True:
            end = mm.find(b'\n', pos)
            if end == -1:
                line = mm[pos:]
                if line:
                    yield json.loads(line.decode('utf-8').strip())
                break
            line = mm[pos:end]
            if line:
                yield json.loads(line.decode('utf-8').strip())
            pos = end + 1
        mm.close()

for obj in fast_read_jsonl('large.jsonl'):
    # быстрый потоковый доступ
    pass

(метод полезен для очень больших файлов, снижает нагрузку на копирование строк в память)

7. Построчная запись с атомарностью через временный файл

import json
import tempfile
import os

def safe_write_jsonl(filename, records):
    tmp = filename + '.tmp'
    with open(tmp, 'w', encoding='utf-8') as f:
        for rec in records:
            f.write(json.dumps(rec, ensure_ascii=False) + '\n')
    os.replace(tmp, filename)

safe_write_jsonl('important.jsonl', [{'x': 1}, {'y': 2}])
print("Записано атомарно")