Apache Kafka

Kafka — распределённая платформа потоковых событий: миллионы сообщений в секунду, воспроизведение логов событий и надёжная доставка для микросервисов и big data.

Уровень: средний 19 разделов ⏱ 3–4 часа

1. Что такое Kafka

Kafka хранит события (сообщения) в топиках. Продюсеры пишут события, консьюмеры читают их. Данные не удаляются сразу — хранятся определённое время (retention), поэтому события можно перечитывать.

Архитектурно: брокеры (серверы), ZooKeeper или KRaft (имплементация контроль-плейна).

2. Установка

docker run -d --name kafka -p 9092:9092 apache/kafka

docker exec -it kafka /opt/kafka/bin/kafka-topics.sh \
    --bootstrap-server localhost:9092 --list

3. Топики и партиции

kafka-topics.sh --bootstrap-server localhost:9092 \
    --create --topic orders --partitions 3 --replication-factor 1

kafka-topics.sh --bootstrap-server localhost:9092 \
    --describe --topic orders

Партиция — упорядоченная очередь сообщений; параллелизм растёт с их числом.
Порядок гарантирован внутри партиции, не между ними.
Replication-factor 3 в проде — реплики на других брокерах.

4. Продюсер

from kafka import KafkaProducer

producer = KafkaProducer(
    bootstrap_servers='localhost:9092',
    value_serializer=lambda v: v.encode())

producer.send('orders', value='{"user": 1, "sum": 99}')
producer.flush()

Ключ сообщения определяет партицию: producer.send('t', key=b'user-1', value=...).

5. Консьюмер и группы

from kafka import KafkaConsumer

consumer = KafkaConsumer(
    'orders',
    bootstrap_servers='localhost:9092',
    group_id='etl',
    auto_offset_reset='earliest')

for msg in consumer:
    print(msg.key, msg.value)

События в группе распределяются между участниками: пока жив group_id, смещения (offsets) сохраняются.

Новая группа читает с начала (earliest) или с конца (latest) — управляется параметром auto_offset_reset.

6. Offset и rebalance

Консьюмер коммитит смещение после обработки. Rebalance происходит, когда участник группы уходит/приходит — партиции перераспределяются. Для строгой обработки используй commit после успеха (enable_auto_commit=False).

7. Гарантии доставки

Режим	Гарантия
at-most-once	не более одного раза (быстро)
at-least-once	не менее одного раза (по умолчанию)
exactly-once	ровно один раз (транзакции/idempotent)

Продюсер с enable_idempotence=True + чтение и запись в транзакциях дают exactly-once.

8. Соединение с другими системами

Kafka Connect: коннекторы в ClickHouse, S3, БД.
kafka.sink в движок для загрузки в хранилища.
Streams/Kafka Streams и Spark Structured Streaming — обработка на лету.

9. Консьюмер в микросервисах

Типичный сценарий: заказы пишутся в топик; микросервисы «оплата», «доставка», «аналитика» читают независимо (разные group_id) — слабо связанная событийная архитектура без прямой связи сервисов.

10. Мониторинг

Метрики: lag консьюмеров (kafka-consumer-groups.sh --describe).
Grafana + Prometheus: экспортер kafka-exporter.
kafka-ui / AKHQ — веб-интерфейсы топиков.

11. Лучшие практики

Именуй топики иерархично: domain.event.v1.
Число партиций под планируемый параллелизм (не меняется без пересоздания).
Не читай события «напрямую» в REST-респонс — только асинхронно.

Партиций меньше, чем консьюмеров в группе — лишние простаивают. Больше партиций = больше параллелизма, но и оверхеда на ребаланс.

12. Schema Registry и Avro

Значения в топике — просто байты. Schema Registry хранит схемы сообщений и выдаёт их продюсерам и консьюмерам, гарантируя совместимость версий:

curl -X POST http://schema-registry:8082/subjects/events-value/versions \
     -H "Content-Type: application/vnd.schemaregistry.v1+json" \
     --data '{"schema": "{\"type\":\"record\",\"name\":\"Event\",\"fields\":[{\"name\":\"id\",\"type\":\"string\"}]}"}'

Avro — бинарный формат с эволюцией схемы.
BACKWARD — консьюмер старой версии читает новые сообщения.
FORWARD — новый консьюмер читает старые данные.

Сами данные в топик не способны рассказать о себе — схема это контракт. Держи Registry рядом с кластером и делай его частью пайплайна деплоя.

13. Kafka Streams

Библиотека Kafka Streams обрабатывает события «на лету», не требуя отдельного кластера поверх топиков:

KStream<String, Order> orders = builder.stream("orders");
orders.filter((key, order) -> order.getAmount() > 1000)
      .to("big-orders");

Цепочка операторов — это топология: map, groupBy, aggregate и оконные функции привносят вычисления без внешних средств:

Stateful-операции хранят состояние в локальном RocksDB-кэше.
Changelog-топики восстанавливают состояние после сбоя.
Параллелизм задаётся числом тасков по партициям.

14. Точно-один раз и идемпотентность

Ретрай на сети дают дубликаты. Kafka решает это на двух уровнях: идемпотентный продюсер (enable.idempotence=true) исключает дубли при ретраях записей, а транзакции связывают записи и offset в одно атомарное действие:

props.put("enable.idempotence", "true");
props.put("transactional.id", "txn-orders-01");

Комбинация даёт гарантию exactly-once: сообщение из топика не потеряется и не обработается дважды, даже если консьюмер упадёт на середине.

Транзакции стоят производительности: читай только нужный диапазон partition, а commit interval поднимай, когда требуется высокая пропускная способность ценой задержки видимости.

15. Типовые ошибки и диагностика

Симптом	Наиболее частая причина
OffsetOutOfRangeException	Retention удалил старые сообщения, а консьюмер отстал
Частые rebalance	Слишком маленький heartbeat.interval в группе
UnknownTopicOrPartition	Опция auto.create выключена или ошибка в имени топика
Медленная запись	acks=all при недоступности min.insync.replicas

Начинай диагностику с команды kafka-consumer-groups.sh --describe --group my-group: разница между LAG и текущим offset показывает, где данные застревают.

Мониторинг продакшена реально держи на метриках кластера: under-replicated partitions и зелёные консьюмеры — первые признаки здоровья Kafka.

16. Ретеншн и очистка логов

Топики копят данные бесконечно, поэтому кластеру нужна политика хранения. Kafka удаляет старые сообщения по времени или размеру сегментов:

retention.ms=604800000        # хранить 7 дней
retention.bytes=1073741824    # или максимум 1 ГБ на партицию
cleanup.policy=delete         # бросаем старые сообщения

По умолчанию клинер срабатывает периодически, а не мгновенно после настройки.
Для дедупликации и хранения «последнего значения» ключа используют cleanup.policy=compact: сохраняется только свежий state по ключу.
Компация требует топики, где сообщения несут ключ — это база для changelog в Kafka Streams.

Ретеншн — это гарантия ограниченного времени жизни, а не бэкап. Удалённые данные восстановить нельзя: планируй retention вместе с консьюмерами, чтобы никто не «отстал» за грань очистки.

17. Зеркалирование и передача в другие кластеры

Миграция на новые брокеры, резервный центр или анализ в облаке требует переноса топиков. Инструмент — MirrorMaker или коннекторы Kafka Connect:

kafka-mirror-maker.sh --consumer.config consumer.properties \
    --producer.config producer.properties \
    --whitelist "orders.*"

Топики-источники читаются, сообщения пишутся в целевой кластер с префиксом имени источника по умолчанию.
Offset консьюмеров переносятся отдельно: --consumer-offset-sync помогает продолжить без потерь.
Проще для точечной синхронизации — Kafka Connect с коннекторами S3, JDBC и Elasticsearch: данные уходят стримингом, не превращая кластер в витрину для копирования.

Полная репликация не автоматическая: MirrorMaker дублирует топики по whitelist, удалённые события в целевом не гаснут сами — планируй ретеншн и очистку отдельно.

18. Тюнинг производительности и подводные камни

Kafka быстрый, но не магический. Основные рычаги — батчинг, компрессия и правильные настройки чтения:

props.put("batch.size", 32768);       // большие батчи — меньше сетевых проходов
props.put("linger.ms", 5);            // короткая пауза для сборки батча
props.put("compression.type", "lz4"); // экономия полосы на больших объёмах
props.put("max.poll.records", 500);   // контроль темпа консьюмера

Право выбора между linger.ms и задержкой: 5 мс почти не заметны для пользователя, но радикально поднимают throughput.
Компрессия lz4/zstd выгоднее zlib на скорости; на текстовых JSON — экономия 2–4 раза.
Один топик — один типичный размер сообщения: смешивать 1 КБ и 10 МБ в одном топике плохо для сегментов и кэша.
Частый подводный камень — fetch.min.bytes и слишком малый max.poll.interval.ms: медленная обработка заставляет группу «выпадать» и перебрасывать партиции.

Перед настройками измерь базовый лаг: тюнинг без диагностики — гадание. Смотри kafka-consumer-groups.sh --describe и только потом меняй параметры.

19. Безопасность: TLS, SASL и ACL

Прод-кластер с открытыми портами — готовый сюрприз для атакующих. Три уровня защиты:

listeners=SSL://:9093,SASL_PLAINTEXT://:9094
ssl.truststore.location=/etc/kafka/truststore.jks
sasl.enabled.mechanisms=SCRAM-SHA-256

TLS шифрует трафик между клиентом и брокером — обязателен на незащищённой сети.
SASL — аутентификация клиентов: SCRAM (парольная), Kerberos (корпоративное SSO), PLAIN (простая, обычно поверх TLS).
ACL — авторизация: кому что можно. Принцип минимума прав: консьюмер только читает свой топик, продюсер пишет свой.

kafka-acls.sh --bootstrap-server localhost:9093 \
    --add --allow-principal User:app1 --operation Read --topic orders \
    --group analytics

Если включаешь authorizer.class.name — по умолчанию всё запрещено. Распиши ACL заранее и проверь доступ суперпользователя, иначе кластер «ослепнет» сразу после перезапуска.

20. Локальная разработка и тесты

Полный классический кластер — тяжёлый для быстрых итераций. Для разработки используйте контейнерные варианты и заготовленные окружения.

docker run -p 9092:9092 --rm apache/kafka-native:latest

Testcontainers поднимает брокер на время модульного теста продюсеров и консьюмеров и удаляет его после запуска.
Веб-интерфейсы (kafka-ui, Kafdrop) помогают инспектировать топики и офсеты без CLI.
Для эмуляции сбоев используйте toxiproxy: обрыв сети, задержки, потерянные пакеты.

Держите профиль local с маленькими таймингами ретеншена и одним брокером: прогоны быстрее, а картина ошибок нагляднее.

21. Quotas и защита от перегрузки

Один прожорливый клиент может «съесть» весь кластер. Квоты ограничивают клиентов на уровне брокера, а не надежда на их воспитанность.

Квота	Что ограничивает
Network bytes	входящий/исходящий трафик клиента
CPU quota	потолок использования CPU на клиента
Request rate	частота запросов к брокеру

Квоты задаются через kafka-configs --alter --add-config и объединяются в группы клиентов.

Квоты не заменяют мониторинг: следите за метриками kafka_server_quota_* в Grafana, чтобы увидеть троттлинг до того, как он ударит по продюсерам.

22. Kafka в Kubernetes и KRaft

Экосистема уходит от ZooKeeper, а управление брокерами в Kubernetes требует декларативного подхода.

Strimzi (оператор) создаёт кластер из CRD: cluster, topics, users — инфраструктура становится кодом.
KRaft убирает ZooKeeper: контроллер встроен в кластер, меньше компонентов и точек отказа.
Redpanda — совместимая альтернатива на C++, проще разворачивается в стейтфул-наборах.

Для тестовой среды используйте kraft-combined и один под с PVC: быстрый старт и простые бэкапы через snapshot тома.

Перед продакшеном проверьте: распределение по зонам, минимум три реплики контроллеров, отдельные диски под логи.

23. Планирование ёмкости

Kafka не растёт сама: партиции, реплики и диски рассчитываются заранее, а не после падения брокера.

Партиция — единица распараллеливания: сотни партиций на одном брокере повышают нагрузку на лидеров и фолловеров.
Репликация 3x умножает объём хранения и сетевой трафик между брокерами.
Сегменты и log.retention.* определяют, сколько диска нужно под историю событий.

Параметр	Типичная оценка
Партиций на топик	кратно числу консьюмеров в группе
Реплики	минимум 3
Диск	запись в сутки × дни хранения × 2 на пики и ребалансинги

Не увеличивайте ретеншен бездумно: посчитайте скорость записи, умножьте на дни хранения и проверьте диск под пиковую нагрузку и ребалансинги.

Что дальше

Объедини Kafka с Spark и ClickHouse для полного data-пайплайна.

← Вернуться к курсам