ITcompressor IT — сжато и по делу
← Назад к статьям

Apache Kafka

10.06.2023
kafkastreamingbig data

Apache Kafka — распределённая потоковая платформа: продюсеры пишут события в топики, консьюмеры группы читают их независимо. Топик делится на партиции — единицы параллелизма и порядка (сообщения в пределах партиции упорядочены).

Все примеры дальше — вокруг одного топика orders-events с тремя партициями на локальном брокере, кластер — в отдельном разделе.

Установка#

Kafka — JVM-приложение: нужен Java 17. Свежую версию и ссылку на архив — проверяйте на kafka.apache.org/downloads, номера версий в URL ниже подставьте под актуальную.

# Зависимости
sudo apt update
sudo apt install openjdk-17-jdk wget

# Скачивание и распаковка (Kafka 3.9, сборка под Scala 2.13)
wget https://downloads.apache.org/kafka/3.9.0/kafka_2.13-3.9.0.tgz
tar -xzf kafka_2.13-3.9.0.tgz
sudo mv kafka_2.13-3.9.0 /opt/kafka

Запуск (KRaft)#

С версии 3.3 Kafka работает в режиме KRaft — метаданные кластера хранятся в Raft-логе самой Kafka, ZooKeeper не нужен (в 4.0 он удалён полностью).

cd /opt/kafka

# Сгенерировать Cluster ID кластера
KAFKA_CLUSTER_ID="$(bin/kafka-storage.sh random-uuid)"

# Отформатировать директорию метаданных и логов
bin/kafka-storage.sh format -t "$KAFKA_CLUSTER_ID" -c config/kraft/server.properties

# Старт брокера (-daemon — фоновый процесс, лог в logs/server.log)
bin/kafka-server-start.sh -daemon config/kraft/server.properties

# Проверка живости
tail -f /opt/kafka/logs/server.log

Kafka 3.x ещё умеет классический режим с ZooKeeper (bin/zookeeper-server-start.sh + config/zookeeper.properties), но новые развёртывания имеет смысл делать только в KRaft.

Топики#

# Создать топик: 3 партиции, одна копия данных (одиночный брокер)
bin/kafka-topics.sh --create --topic orders-events \
  --bootstrap-server localhost:9092 --partitions 3 --replication-factor 1

# Список топиков
bin/kafka-topics.sh --list --bootstrap-server localhost:9092

# Описание: партиции, лидеры, офсеты
bin/kafka-topics.sh --describe --topic orders-events --bootstrap-server localhost:9092

--partitions — сколько параллельных потоков чтения и записи (после создания только увеличивают). --replication-factor 1 — единственный вариант для одиночного брокера; в кластере ставят 2–3.

Производители и потребители#

Производитель#

Сообщения пишутся построчно со stdin:

bin/kafka-console-producer.sh --topic orders-events --bootstrap-server localhost:9092
> {"order_id": 1001, "amount": 149.99}
> {"order_id": 1002, "amount": 59.50}

Потребитель#

# Читать с начала истории (--from-beginning), иначе — только новые
bin/kafka-console-consumer.sh --topic orders-events \
  --bootstrap-server localhost:9092 --from-beginning

Группы потребителей#

Консьюмеры одной группы делят партиции между собой: каждая партиция закрепляется ровно за одним потребителем группы — так Kafka масштабирует обработку, не теряя порядка.

# Потребитель из группы billing
bin/kafka-console-consumer.sh --topic orders-events \
  --bootstrap-server localhost:9092 --group billing

# Состояние группы: закрепления, текущие офсеты, отставание (lag)
bin/kafka-consumer-groups.sh --bootstrap-server localhost:9092 \
  --describe --group billing

Офсеты группы Kafka хранит сама (в топике __consumer_offsets) — перезапущенный потребитель продолжает с места остановки.

Кластер из нескольких брокеров#

В KRaft узлы совмещают роли брокера и контроллера и сами выбирают «мета»-лидера. Типичной прод-схемой являются три узла kafka1–kafka3: роли, узлы и кворум контроллеров описываются в server.properties каждого узла:

# config/kraft/server.properties, узел kafka1 (на kafka2/kafka3 — node.id 2 и 3)
process.roles=broker,controller
node.id=1
controller.quorum.voters=1@kafka1.example.com:9093,2@kafka2.example.com:9093,3@kafka3.example.com:9093
listeners=PLAINTEXT://:9092,CONTROLLER://:9093
advertised.listeners=PLAINTEXT://kafka1.example.com:9092
log.dirs=/var/kafka-logs
num.partitions=3

На каждом узле — свой node.id, но один и тот же controller.quorum.voters и один Cluster ID при форматировании:

bin/kafka-storage.sh format -t "$KAFKA_CLUSTER_ID" -c config/kraft/server.properties
bin/kafka-server-start.sh -daemon config/kraft/server.properties

В кластере создавайте топики с --replication-factor 2–3: при падении одного брокера партиция останется читаемой и записываемой за счёт копии на другом узле.

Брандмауэр#

# Клиентам — консумерский/продюсерский порт
sudo ufw allow from 10.0.0.0/24 to any port 9092

# Между узлами кластера — контроллерный порт (перекрывающие подсети допустимы)
sudo ufw allow from 10.0.0.0/24 to any port 9093

Порт 2181 (ZooKeeper) больше не участвует в KRaft-развёртывании.

← Назад к статьям