Разворачиваем Apache Kafka

Моя цель - предложение широкого ассортимента товаров и услуг на постоянно высоком качестве обслуживания по самым выгодным ценам.

Прежде чем перейти к статье, хочу вам представить, экономическую онлайн игру Brave Knights, в которой вы можете играть и зарабатывать. Регистируйтесь, играйте и зарабатывайте!

В предыдущей статье мы достаточно подробно рассмотрели архитектуру Apache Kafka, из каких компонентов состоит данное решение, что для чего предназначено. И в этой статье мы рассмотрим процесс развертывания данного решения.

Для запуска Kafka можно использовать решения для управления кластером, такие как ZooKeeper или KRaft. В статье мы рассмотрим работу с ZooKeeper. Zookeper это распределенное приложение для управления кластером, состоящим из большого количества узлов.

Перед установкой Zookeeper не забудьте поставить Java, например вот так

apt -y install openjdk-8-jre

Для установки zookeeper сначала перейдем на официальную страницу данного решения https://zookeeper.apache.org/releases.html и выберем последнюю версию.

Загрузим данную версию с помощью wget

wget https://dlcdn.apache.org/zookeeper/zookeeper-3.8.1/apache-zookeeper-3.8.1-bin.tar.gz

Далее создадим каталог и распакуем в него архив:

mkdir /opt/zookeeper

tar zxvf apache-zookeeper-3.8.1.tar.gz -C /opt/zookeeper --strip-components=1

Для работы нам также потребуются дополнительные каталоги

mkdir -p /opt/zookeeper/data /var/log/zookeeper

Далее мы переходим к конфигурированию системы. Для этого нам потребуется файл /opt/zookeeper/conf/zoo.cfg.

Этот файл необходимо наполнить следующим содержимым:

tickTime = 2000

maxSessionTimeout = 50000

syncLimit = 5

initLimit = 300

autopurge.purgeInterval = 1

autopurge.snapRetainCount = 5

snapCount = 200000

clientPort = 2181

maxClientCnxns = 100

4lw.commands.whitelist=stat

dataDir = /opt/zookeeper/data

dataLogDir = /var/log/zookeeper

 

Мы не будем в этой статье подробно останавливаться на значении каждого из параметров, так как это несколько выходит за рамки статьи.

Далее мы настроим запуск приложения в качестве сервиса. Для этого создадим специальную учетку, настроим ей права и создадим файл юнита в systemd.

useradd -r -c 'Zookeeper service' zookeeper

chown -R zookeeper:zookeeper /opt/zookeeper /var/log/zookeeper

Файл юнита /etc/systemd/system/zookeeper.service будет иметь следующую структуру:

[Unit]

Description=ZooKeeper Service

Documentation=https://zookeeper.apache.org/

Requires=network.target

After=network.target

 

[Service]

Type=forking

User=zookeeper

Group=zookeeper

WorkingDirectory=/opt/zookeeper

ExecStart=/opt/zookeeper/bin/zkServer.sh start

ExecStop=/opt/zookeeper/bin/zkServer.sh stop

ExecReload=/opt/zookeeperbin/zkServer.sh restart

TimeoutSec=30

Restart=on-failure

 

[Install]

WantedBy=default.target

 

Затем нам необходимо просто перезапустить system и запустить zookeper

systemctl daemon-reload

systemctl enable zookeeper --now

Проверим статус zookepeer

systemctl status zookeeper

Базовые настройки Zookeeper мы выполнили. Настройка управления отдельными узлами кластера опять таки не входит в тему данной статьи, поэтому далее мы перейдем к установке Kafka.

Ставим Кафку

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

$ tar -xzf kafka_2.13-3.4.0.tgz

$ cd kafka_2.13-3.4.0

Далее нам необходимо запустить Zookeper с настройками, необходимыми для работы Kafka.

$ bin/zookeeper-server-start.sh config/zookeeper.properties

При успешном запуске мы получим примерно такую "портянку".

Далее откроем еще одну SSH сессию в которой собственно запустим Kafka

# Start the Kafka broker service

$ bin/kafka-server-start.sh config/server.properties

Здесь нас никакой особой псевдографикой не порадовали, так что просто смотрим вывод на предмет ошибок.

Как видно, мы создали базовую конфигурацию для работы Kafka. Для того, чтобы не плодить окна командной строки или SSH сессии я бы посоветовал перезапустить две последние команды в фоновом режиме, т.е. добавить &.

Создаем топик

Напомню из материалов предыдущей статьи, что Kafka - это платформа распределенной потоковой передачи событий, которая позволяет вам читать, записывать, хранить и обрабатывать сообщения на многих узлах.

Для начала в качестве примера создадим топик quickstart-events с помощью команды:

$ bin/kafka-topics.sh --create --topic quickstart-events --bootstrap-server localhost:9092

Давайте посмотрим статистику по созданному топику:

$ bin/kafka-topics.sh --describe --topic quickstart-events --bootstrap-server localhost:9092

Как видно, пока наш топик пуст и самое время начать записывать в него сообщения. Для записи сообщений в топик quickstart-events воспользуемся следующей командой:

$ bin/kafka-console-producer.sh --topic quickstart-events --bootstrap-server localhost:9092

После запуска в интерактивном режиме вводим строки (Ctrl+C для завершения):

Как видите, я не переключил кодировку в первой строке и отправил строку на кириллице. Посмотрим как ее сохранит Kafka.

Для чтения событий из топика quickstart-events применим команду:

$ bin/kafka-console-consumer.sh --topic quickstart-events --from-beginning --bootstrap-server localhost:9092

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

Подключаем источники

Естественно, представленный вариант работы с сообщениями “вручную” на практике мало применим. Гораздо интереснее попробовать в автоматическом режиме прочитать сообщения из источника и поместить их в топик Kafka. В качестве примера мы импортируем данные из файла в раздел Kafka и затем экспортируем данные из раздела Kafka в файл.

Для этого нам потребуется указать в свойствах plugin.path файла config/connect-standalone.properties файл connect-file-3.4.0.jar. Откроем это файл и добавим в него строку

plugin.path=libs/connect-file-3.4.0.jar

Собственно все, далее создаем текстовый файл с сообщениями, которые мы будем читать

echo -e "event1\nevent2\nevent3" > test.txt

Далее нам необходимо запустить коннектор, который будет читать сообщения из этого файла и коннектор, который будет сохранять сообщения в sink файле. В наборе файлов, идущих вместе с Kafka есть файлы config/connect-file-source.properties и config/connect-file-sink.properties. В первом файле указаны настройки для считывания файла test txt, а втором настройки для файла хранения сообщений test.sink.txt. Когда команда ниже будет выполнена коннектор, считывающий сообщения из исходного файла test.txt будет загружать их в топик connect-test (который также есть в конфигурационных файлах), а sink коннектор начнет читать сообщения из топика connect-test и запишет их в файл test.sink.txt.

bin/connect-standalone.sh config/connect-standalone.properties config/connect-file-source.properties config/connect-file-sink.properties

Убедиться в работе коннектора можно с помощью просмотра содержимого test.sink.txt

Заключение

На этом, я полагаю сегодняшнюю статью можно завершить. Мы развернули Apache Kafka и Zookeeper, посмотрели как можно вручную создавать топики, добавлять и читать сообщения, а также развернули коннектор, читающий данные из текстового файла. В следующей статье мы продолжим изучение практических аспектов работы с Kafka. А сейчас хочу пригласить вас на бесплатный урок, в рамках которого рассмотрим как в приложениях на Spring Boot можно работать с Kafka. Узнаем, что предоставляет платформа Spring для ускоренной разработки приложений, работающих с Kafka. Посмотрим, какие есть настройки, как это все конфигурируется. Проведем границу между "родным функционалом" Kafka api и "добавками" от Spring Boot.

  • Зарегистрироваться на бесплатный урок

Источник: https://habr.com/ru/companies/otus/articles/727498/


Интересные статьи

Интересные статьи

Цепочка поставок в пищевой промышленности и ритейле — это сложная, медленная и ненадёжная система. В этой статье мы рассмотрим развёртывание Apache Kafka для обработки данных в реальном времени в...
В этой статье показано, как можно визуализировать Apache Kafka Streams с реактивными приложениями с помощью пользовательского интерфейса разработчика в Quarkus (Quarkus Dev UI). Quarkus...
Поймал как-то программист золотую рыбку, а она ему и говорит человеческим голосом: «Отпусти меня в синее море, я исполню любое твоё желание». Программист ЦПУшечкой поскрипел, да и выдал техзадание. «П...
Открываем предзаказ продвинутого курса по Apache Kafka. Видеокурс о том, как настроить и оптимизировать Apache Kafka — брокер сообщений для микросервисов. Вы последовательно узнаете, ...
Реализация ORM в ядре D7 — очередная интересная, перспективная, но как обычно плохо документированная разработка от 1с-Битрикс :) Призвана она абстрагировать разработчика от механики работы с табл...