Apache Avro: row-oriented бинарный формат с эволюцией схемы

JSON и CSV — текстовые форматы. Удобно человеку, неэффективно машине: имена полей повторяются в каждой записи, числа кодируются как строки, парсинг медленный. Бинарные форматы решают эти проблемы — но платят сложностью и необходимостью схемы.

Apache Avro — бинарный row-oriented формат, родившийся в проекте Hadoop в 2009 году. Сегодня это стандартный сериализатор для Apache Kafka (через Confluent Schema Registry), популярный выбор для долгосрочного хранения и инкрементальной выгрузки.

Что значит row-oriented

Avro хранит записи последовательно: запись 1, запись 2, запись 3… Каждая запись содержит все свои поля рядом. Это противоположность columnar-формату Parquet, где значения одной колонки лежат рядом, а полная запись разбросана по разным местам.

Поэтому Avro доминирует в Kafka (одна запись = одно сообщение), а Parquet — в data lake-е (миллионы записей, аналитика).

Schema в JSON

Сердце Avro — схема, описанная в JSON. Без схемы данные нельзя ни сериализовать, ни прочитать.

{
  "type": "record",
  "name": "User",
  "namespace": "com.example.events",
  "fields": [
    {"name": "id", "type": "long"},
    {"name": "email", "type": "string"},
    {"name": "name", "type": ["null", "string"], "default": null},
    {"name": "age", "type": "int", "default": 0},
    {"name": "tags", "type": {"type": "array", "items": "string"}, "default": []},
    {"name": "status", "type": {
        "type": "enum",
        "name": "UserStatus",
        "symbols": ["ACTIVE", "INACTIVE", "BANNED"]
    }, "default": "ACTIVE"}
  ]
}

Несколько ключевых моментов:

• Тип записи (record) с именем и пространством имён.
• Поля — упорядоченный массив. Порядок важен: бинарный формат пишет значения в порядке schema, без имён полей.
• Union-тип ["null", "string"] — стандартный способ сделать поле nullable.
• Default — обязателен для optional полей при schema evolution.

Типы Avro

Primitive:

Complex:

• record — структура с именованными полями.
• enum — фиксированный набор символов.
• array — повторяющиеся значения одного типа.
• map — словарь со строковыми ключами.
• union — одно значение из нескольких возможных типов (["null", "string"]).
• fixed — массив байт фиксированной длины (например, для UUID).

В этом и смысл schema-driven формата: имена полей не передаются. Они известны из schema. Передаются только значения, в строго заданном порядке.

Schema evolution

Главная фишка Avro — продуманная эволюция схемы. В реальности схема меняется: добавляются поля, убираются устаревшие. Avro позволяет читать данные, записанные другой версией схемы, благодаря разделению writer’s schema (с какой записывалось) и reader’s schema (с какой читаем).

Совместимость бывает разной:

• Backward compatible — новый reader умеет читать старые данные. Самое частое требование при добавлении поля.
• Forward compatible — старый reader умеет читать новые данные. Полезно, когда продьюсер обновляется быстрее консьюмеров.
• Full compatible — оба направления одновременно.
• Breaking — никто никого не понимает.

Что можно делать без поломки backward-совместимости:

1. Добавить поле с default — старые данные читаются с дефолтом.
2. Удалить поле, имеющее default — старые данные игнорируют поле.
3. Расширить enum новыми символами (если есть default).
4. Расширить union новыми типами в начало или в конец.

Что нельзя без breaking change:

• Поменять тип поля с int на string.
• Добавить required-поле без default.
• Удалить required-поле, у которого нет default.
• Переименовать поле без aliases.

// Versions evolution
// v1
{"name": "name", "type": "string"}
// v2 -- добавляем optional email
{"name": "email", "type": ["null", "string"], "default": null}
// Backward compatible: новый reader читает v1 -> email = null
// Forward compatible: старый reader читает v2 -> игнорирует поле email

fastavro в Python

fastavro — самая быстрая Python-библиотека (на C-расширениях, без Java JVM). Стандартная в DE-pipeline-ах.

import fastavro
from io import BytesIO

schema = {
    "type": "record",
    "name": "Order",
    "namespace": "com.shop.events",
    "fields": [
        {"name": "id", "type": "long"},
        {"name": "user_id", "type": "long"},
        {"name": "total", "type": "double"},
        {"name": "currency", "type": {"type": "enum", "name": "Currency", "symbols": ["USD", "EUR", "RUB"]}},
        {"name": "items", "type": {"type": "array", "items": "string"}, "default": []},
    ],
}
parsed = fastavro.parse_schema(schema)

records = [
    {"id": 1, "user_id": 100, "total": 49.90, "currency": "USD", "items": ["sku-1", "sku-2"]},
    {"id": 2, "user_id": 100, "total": 12.50, "currency": "EUR", "items": ["sku-9"]},
]

# Сериализация в container file
with open("orders.avro", "wb") as out:
    fastavro.writer(out, parsed, records, codec="snappy")

# Чтение
with open("orders.avro", "rb") as inp:
    for record in fastavro.reader(inp):
        print(record)

Для сериализации одной записи без контейнера (например, для Kafka):

buf = BytesIO()
fastavro.schemaless_writer(buf, parsed, records[0])
binary_message = buf.getvalue()  # отправляем в Kafka

И обратно:

buf = BytesIO(binary_message)
record = fastavro.schemaless_reader(buf, parsed)

Container файлы

Avro-файл (.avro) — это контейнер со встроенной схемой. Структура:

+------------------------------+
| Magic bytes 'Obj' + version  |  4 байта
+------------------------------+
| File metadata (Map<string,bytes>) |
|   - avro.schema = JSON-schema    |
|   - avro.codec  = "snappy"        |
+------------------------------+
| Sync marker (16 random bytes) |  фиксированный для всего файла
+------------------------------+
| Block 1                       |
|   - count (varint)            |
|   - size (varint)             |
|   - данные (опц. сжатые)      |
|   - sync marker               |
+------------------------------+
| Block 2                       |
| ...                           |
+------------------------------+

Sync marker позволяет читать файл с произвольного места — нужно найти ближайший marker. Так Hadoop / Spark делят файл на splits для параллельной обработки.

Avro в Kafka и Schema Registry

Прямое использование Avro в Kafka даёт проблему: каждое сообщение должно содержать или ссылаться на schema. Решение — Confluent Schema Registry:

1. Producer регистрирует schema в Registry, получает schema_id (4 байта).
2. Producer пишет в Kafka: [magic byte 0x00] [schema_id 4 байта] [Avro binary payload].
3. Consumer читает первые 5 байт, по schema_id достаёт schema из Registry, десериализует payload.

• Централизованное хранение всех схем кластера.
• Compatibility-проверки на этапе регистрации (нельзя выкатить breaking change случайно).
• Кэширование на стороне consumer-а (один HTTP-запрос на schema_id, потом из памяти).
• Версионирование — каждая schema получает новый ID при изменении.

Альтернативы Confluent Registry: AWS Glue Schema Registry, Apicurio (open-source).

Avro vs Protobuf

И тот, и другой — schema-first бинарные форматы. Сравнение:

Правило: Avro — для долгосрочного storage и streaming с эволюцией схемы. Protobuf — для inter-service RPC через gRPC. Не конкуренты, а инструменты для разных задач.

Сжатие

Avro поддерживает несколько кодеков сжатия per-block:

• null — без сжатия. Быстрее всего, больше всего места.
• deflate — стандартное zlib-сжатие. Универсальный baseline.
• snappy — Google Snappy. В 2-3 раза быстрее, чем deflate, чуть хуже compression ratio. Стандарт в data engineering.
• zstandard — современный кодек от Facebook. Лучший trade-off скорость/размер.
• bzip2 — медленный, но компактный. Редко.

Snappy — дефолтный выбор для Kafka и data lake-ов.

Где DE встретит Avro

• Kafka — почти любой production-Kafka использует Avro + Schema Registry.
• Hadoop / HDFS — традиционное хранение событийных данных.
• Spark — нативная поддержка через spark-avro.
• CDC-инструменты — Debezium шлёт изменения в БД как Avro-сообщения.
• Confluent Stream Lineage — построение data lineage по schema-метаданным.

Junior DE в первую неделю работы со Kafka столкнётся с Avro: производитель данных, consumer данных или просто пытается понять, что лежит в топике.