JSONL и MessagePack: следующий шаг после JSON

Когда обычный JSON становится узким местом, есть два направления развития. Первое — JSONL (JSON Lines, NDJSON): один JSON-объект на строку, текстовый формат, который позволяет потоковую обработку. Второе — MessagePack: бинарный формат, совместимый с JSON по semantically, но компактнее на 30-50% и быстрее в 3-5 раз.

Для data engineer оба важны: JSONL — стандарт для логов и event streams, MessagePack — для inter-service communication и кэширования. В конце урока — таблица, когда выбирать какой формат: JSON, JSONL, MessagePack, Protobuf, Avro, Parquet.

JSONL: одна запись на строку

JSONL (JSON Lines, иногда NDJSON — Newline-Delimited JSON) — это не новый формат, а соглашение о расположении JSON-данных. Файл состоит из строк, каждая строка — валидный JSON-объект. Между объектами — \n, никаких других разделителей.

{"id": 1, "name": "Alice", "age": 30}
{"id": 2, "name": "Bob", "age": 25}
{"id": 3, "name": "Carol", "age": 35}

Это не валидный JSON — JSON-парсер на весь файл упадёт (несколько корневых элементов — невалидно). Парсить нужно построчно.

Преимущества JSONL над monolithic JSON:

• Streaming: можно обрабатывать построчно. Файл 100 ГБ без памяти 100 ГБ.
• Append-only: добавить запись = f.write(json.dumps(record) + '\n'). Не нужно читать-парсить-сериализовать весь файл.
• Resilient: если строка корявая — упала только одна запись, остальные читаются.
• Параллелизм: разные процессы читают разные части файла, разделение по строкам — простое.
• Стандарт для логов: ELK stack, Datadog, AWS CloudWatch, многие BI tools нативно поддерживают.

Недостатки:

• Не валидный JSON — на стороне получателя нужно знать, что это JSONL.
• Каждая запись — independent: нельзя ссылаться между записями.

Парсинг JSONL в Python

Базовый паттерн через генераторы — память O(1):

import json

def read_jsonl(path):
    with open(path, encoding='utf-8') as f:
        for line in f:
            line = line.strip()
            if not line:  # skip empty lines
                continue
            yield json.loads(line)

# Использование
for record in read_jsonl("events.jsonl"):
    process(record)  # обработка по одной записи

# Обработка с фильтрацией
active_users = (r for r in read_jsonl("users.jsonl") if r.get('active'))
emails = (r['email'] for r in active_users)
for email in emails:
    send_notification(email)

Запись:

import json

records = [
    {"id": 1, "name": "Alice"},
    {"id": 2, "name": "Bob"},
    {"id": 3, "name": "Carol"},
]

with open("users.jsonl", "w", encoding="utf-8") as f:
    for r in records:
        f.write(json.dumps(r, ensure_ascii=False) + '\n')

# Append без read whole file
def append_record(path, record):
    with open(path, "a", encoding="utf-8") as f:
        f.write(json.dumps(record, ensure_ascii=False) + '\n')

Обработка ошибок (robust pattern):

import json

def read_jsonl_safe(path):
    with open(path, encoding='utf-8') as f:
        for line_num, line in enumerate(f, 1):
            line = line.strip()
            if not line:
                continue
            try:
                yield json.loads(line)
            except json.JSONDecodeError as e:
                print(f"Bad JSON on line {line_num}: {e}", file=sys.stderr)
                continue  # skip and continue

Pandas умеет читать JSONL:

import pandas as pd
df = pd.read_json("data.jsonl", lines=True)  # lines=True для JSONL!

lines=True — критичный параметр. Без него pandas попробует распарсить весь файл как один JSON и упадёт.

JSONL для логов и event streams

JSONL — стандарт de-facto для structured logs. Пример лога приложения:

{"ts": "2026-05-15T12:30:00Z", "level": "INFO", "service": "api", "msg": "Request received", "request_id": "abc", "method": "GET", "path": "/users"}
{"ts": "2026-05-15T12:30:01Z", "level": "INFO", "service": "api", "msg": "Response sent", "request_id": "abc", "status": 200, "duration_ms": 42}
{"ts": "2026-05-15T12:30:02Z", "level": "ERROR", "service": "worker", "msg": "Failed to process", "task_id": 123, "error": "Connection timeout"}

Преимущества для логов:

• Структурированные поля: легко искать level=ERROR AND service=api.
• Пайплайн обработки: cat logs.jsonl | jq 'select(.level=="ERROR")' | head.
• Импорт в анализ tools: ClickHouse, Snowflake, BigQuery нативно умеют JSONL.

В Python для structured logging — python-json-logger:

import logging
from pythonjsonlogger import jsonlogger

logger = logging.getLogger()
handler = logging.StreamHandler()
handler.setFormatter(jsonlogger.JsonFormatter(
    "%(asctime)s %(levelname)s %(name)s %(message)s",
    rename_fields={"asctime": "ts", "levelname": "level"},
))
logger.addHandler(handler)

logger.info("Request received", extra={"request_id": "abc", "path": "/users"})
# Output: {"ts": "...", "level": "INFO", "name": "...", "message": "Request received", "request_id": "abc", "path": "/users"}

MessagePack: бинарный JSON

MessagePack (msgpack) — бинарный формат с тем же data model, что JSON: те же типы (string, int, float, bool, null, array, map). Но кодируется в bytes, а не в text.

Размер: на 30-50% меньше JSON для типичных данных. Для чисел и булевых — еще больше (true в JSON = 4 байта, в msgpack = 1 байт).

Скорость: парсинг и сериализация в 3-5 раз быстрее JSON (для C-implementation).

# pip install msgpack
import msgpack
import json

data = {"id": 1, "name": "Alice", "scores": [1.5, 2.7, 3.9], "active": True}

# Сериализация
json_bytes = json.dumps(data).encode('utf-8')
msgpack_bytes = msgpack.packb(data)
print(f"JSON: {len(json_bytes)} bytes")
print(f"MessagePack: {len(msgpack_bytes)} bytes")
# JSON: ~70 bytes
# MessagePack: ~40 bytes

# Десериализация
data_back = msgpack.unpackb(msgpack_bytes, raw=False)
print(data_back)
# raw=False -- нужно для строк в str (а не bytes). Старая опция, в новых версиях по умолчанию

# Файлы
with open("data.msgpack", "wb") as f:
    msgpack.pack(data, f)

with open("data.msgpack", "rb") as f:
    loaded = msgpack.unpack(f, raw=False)

# Streaming (как JSONL, но binary)
import msgpack
with open("events.msgpack", "wb") as f:
    for event in events_iter():
        msgpack.pack(event, f)

with open("events.msgpack", "rb") as f:
    unpacker = msgpack.Unpacker(f, raw=False)
    for event in unpacker:
        process(event)

Когда использовать MessagePack:

• Кэш в Redis: значения хранятся как bytes — msgpack экономит память и быстрее.
• Inter-service communication: API между микросервисами, где обе стороны контролируете.
• Файлы для быстрой загрузки: преаггрегированные data, которые часто читаются.
• WebSocket с большим трафиком: меньше байт по сети.

Когда не использовать:

• Public API: HTTP-клиенты ожидают JSON. MessagePack нужно явно поддерживать.
• Debugging: нельзя посмотреть содержимое в text editor. cat msgpack-файла — мусор.
• Когда данные мелкие (< 1 КБ): overhead библиотеки больше выигрыша.

Сравнение всех форматов

Большая таблица, к которой возвращаться.

Полная таблица tradeoffs:

Decision tree:

1. Нужно для public API -> JSON.
2. Нужно для логов / event stream / append-only big data -> JSONL.
3. Нужно для inter-service быстро + компактно, без строгой схемы -> MessagePack.
4. Нужно gRPC / strict-typed API -> Protobuf.
5. Нужно для Kafka + Hadoop/Spark + schema evolution -> Avro.
6. Нужно для analytics на больших data + S3/data lake -> Parquet.

Подводные камли

JSONL и trailing newline. Файл должен заканчиваться \n. Если последняя строка без newline — большинство парсеров справятся, но некоторые tools (jq в строгом режиме) могут потерять последнюю запись.

MessagePack и numpy. msgpack не сериализует numpy arrays/scalars напрямую. Нужно либо tolist() сначала, либо msgpack-numpy extension.

MessagePack и datetime. Базовый msgpack не имеет нативного datetime типа. Используйте default=callable (как в json) для конвертации, или extension (msgpack-extension).

Совместимость версий msgpack. Старые версии (msgpack-python < 1.0) использовали bytes для строк по умолчанию (raw=True). Новые — str (raw=False). При парсинге чужих файлов проверяйте.

Append к JSONL и race conditions. Несколько процессов append’ят в один файл — могут получиться interleaved строки. Решение: использовать atomic file writes или line-buffered I/O с lock.

Попробуй сам

import json
import msgpack
import time

# 1. JSONL streaming
events = [{"id": i, "ts": f"2026-05-15T12:{i:02d}:00Z", "value": i * 1.5} for i in range(10_000)]

# Запись
with open("events.jsonl", "w") as f:
    for e in events:
        f.write(json.dumps(e) + '\n')

# Чтение по строкам
def read_jsonl(path):
    with open(path) as f:
        for line in f:
            line = line.strip()
            if line:
                yield json.loads(line)

count = sum(1 for _ in read_jsonl("events.jsonl"))
print(f"Read {count} events")

# 2. Сравнение размеров JSON vs JSONL vs MessagePack
import os

# Один большой JSON
with open("events.json", "w") as f:
    json.dump(events, f)

# JSONL уже создан выше

# MessagePack
with open("events.msgpack", "wb") as f:
    for e in events:
        msgpack.pack(e, f)

# Один большой msgpack
with open("events_one.msgpack", "wb") as f:
    msgpack.pack(events, f)

for name in ["events.json", "events.jsonl", "events.msgpack", "events_one.msgpack"]:
    print(f"{name}: {os.path.getsize(name):>10} bytes")

# 3. Скорость парсинга
def time_it(label, fn):
    start = time.perf_counter()
    fn()
    print(f"{label}: {time.perf_counter() - start:.3f}s")

time_it("json.load", lambda: json.load(open("events.json")))
time_it("jsonl iterate", lambda: list(read_jsonl("events.jsonl")))
time_it("msgpack.unpack one", lambda: msgpack.unpack(open("events_one.msgpack", "rb"), raw=False))

# 4. MessagePack streaming
def stream_msgpack(path):
    with open(path, "rb") as f:
        unpacker = msgpack.Unpacker(f, raw=False)
        for record in unpacker:
            yield record

for r in stream_msgpack("events.msgpack"):
    if r["id"] < 3:
        print(r)