Установка Kafka на Ubuntu Linux

Apache Kafka - это распределенная канальная платформа. Платформа каналов обладает такими возможностями:

• Публикация и подписка на каналы записей, что очень похоже на очередь сообщений
• Сохранение каналов записей в отказоустойчивом долгосрочном хранилище
• Обработка калаов записей сразу при их появлении

Kafka обычно используется для приложений:

• Построения конвейеров передачи данных в реальном времени между приложениями
• Построения потоковых приложений, преобразовывающих или реагирующих на потоки данных

Концепция:

• Kafka запускается, как кластер, выполняющийся на одном или нескольких серверах, расположенных, возможно, в нескольких дата-центрах
• Kafka сохраняет потоки записей в категориях, называемых "тема"
• Каждая запис состоит из ключа, значения и отметки времени

Kafka содержит четыре API:

• Producer API - позволяет приложениям публиковать потоки записей в одну или несколько тем
• Consumer API - позволяет приложениям подписываться на темы и обрабатывать потоки записей, публикуемых в них
• Streams API - позволяет приложениям работать, как обработчик потоков, предполагая извлечение записей из входящего потока по одной или нескольким темам, и публикуя выходящий поток в одну или несколько тем, выполняя некоторое преобразование входящих потоков в исходящие потоки
• Connector API - позволяет строить и запускать переиспользуемых публикаторов и подписчиков, что соединяет темы Kafka с существующими приложениями. К примеру, коннектор к реляционной базе данных может отлавливать каждое изменение таблицы.

В Kafka коммуникация между клиентами и серверами выполняется на простом, высокопроизводительном языке по TCP протоколу. Протокол версионен и поддерживает обратную совместимость со старыми версиями. Существует множество клиентов для Kafka на различных языках, среди которых Go, Java, PHP, Python, Rust, Swift.

Установка Apache Kafka на Ubuntu Linux

Установить Java

apt install -y default-jre
java -version

Создать пользователя

adduser kafka
adduser kafka sudo
su -l kafka

Установки Apache Kafka 3.1.0 со Scala 2.13

curl -fsSLO https://dlcdn.apache.org/kafka/3.1.0/kafka_2.13-3.1.0.tgz
tar zxvf kafka_2.13-3.1.0.tgz
mv kafka_2.13-3.1.0 ~/kafka
cd ~/kafka
#./bin/kafka-server-start.sh config/server.properties
#./bin/zookeeper-server-start.sh config/zookeeper.properties

Настройка

vi ~/kafka/config/server.properties
# установить следующие значечния
delete.topic.enable = true
log.dirs=/home/kafka/logs

Настроить Systemd Units
sudo vi /etc/systemd/system/zookeeper.service

[Unit]
Requires=network.target remote-fs.target
After=network.target remote-fs.target

[Service]
Type=simple
User=kafka
ExecStart=/home/kafka/kafka/bin/zookeeper-server-start.sh /home/kafka/kafka/config/zookeeper.properties
ExecStop=/home/kafka/kafka/bin/zookeeper-server-stop.sh
Restart=on-abnormal

[Install]
WantedBy=multi-user.target

sudo vi /etc/systemd/system/kafka.service

[Unit]
Requires=zookeeper.service
After=zookeeper.service

[Service]
Type=simple
User=kafka
ExecStart=/bin/sh -c '/home/kafka/kafka/bin/kafka-server-start.sh /home/kafka/kafka/config/server.properties > /home/kafka/kafka/kafka.log 2>&1'
ExecStop=/home/kafka/kafka/bin/kafka-server-stop.sh
Restart=on-abnormal

[Install]
WantedBy=multi-user.target

systemctl start zookeeper
systemctl start kafka
systemctl enable zookeeper
systemctl enable kafka

Тестирование Kafka

# создать тему QUEUE
#./bin/kafka-topics.sh --create --zookeeper localhost:2181 --replication-factor 1 --partitions 1 --topic QUEUE
./bin/kafka-topics.sh --create --bootstrap-server localhost:9092 --replication-factor 1 --partitions 1 --topic QUEUE
# создать публикатора для темы QUEUE, и написать несколько строк сообщений
./bin/kafka-console-producer.sh --broker-list localhost:9092 --topic QUEUE
# показать список тем
#./bin/kafka-topics.sh --list --zookeeper localhost:2181
./bin/kafka-topics.sh --list --bootstrap-server localhost:9092
# создать подписчика на тему QUEUE и просмотреть сообщения
./bin/kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic QUEUE --from-beginning

их соседнего терминала можно отправить сообщение в открытый с подписчиком

echo "Hello World!" | ~/kafka/bin/kafka-console-producer.sh --broker-list localhost:9092 --topic QUEUE > /dev/null

Установка утилиты администрирования KafkaT

sudo apt install ruby ruby-dev build-essential
sudo CFLAGS=-Wno-error=format-overflow gem install kafkat

vi ~/.kafkatcfg

{
  "kafka_path": "~/kafka",
  "log_path": "/home/kafka/logs",
  "zk_path": "localhost:2181"
}

kafkat partitions