Установка Apache Kafka в CentOS 7

Apache Kafka – это популярный распределённый брокер сообщений, предназначенный для обработки больших объемов данных в режиме реального времени. Кластер Kafka обладает не только высокой масштабируемостью и отказоустойчивостью, но также имеет гораздо более высокую пропускную способность по сравнению с другими брокерами сообщений (как ActiveMQ и RabbitMQ). Как правило, Apache Kafka используется в качестве системы обмена сообщениями pub/sub, однако многие организации используют его для логирования, потому что он обеспечивает постоянное хранение опубликованных сообщений.

Система обмена сообщениями pub/sub позволяет публиковать сообщения без учета количества подписчиков или способов их обработки. Подписанные клиенты автоматически уведомляются об обновлениях и появлении новых сообщений. Эта система более эффективна и масштабируема, чем те системы, в которых клиенты проводят периодическую проверку новых сообщений.

В данном мануале вы узнаете, как установить и использовать Apache Kafka 1.1.0 в CentOS 7.

Требования

Для выполнения руководства понадобится:

  • Сервер CentOS 7, настроенный по этому мануалу.
  • 4GB RAM как минимум. Нехватка объема RAM может привести к сбою сервера Kafka, при этом Java virtual machine (JVM) выдает «Out Of Memory» во время запуска.
  • OpenJDK на сервере. Инструкции по установке можно найти в мануале Установка Java на CentOS и Fedora. Kafka написан на Java, поэтому для его работы требуется JVM; однако в его сценарии запуска есть ошибка обнаружения версий, из-за которой он не работает с версиями JVM выше 8.

1: Создание пользователя для Apache Kafka

Apache Kafka может обрабатывать запросы по сети, потому необходимо создать для него специального пользователя. Это сводит к минимуму возможные повреждения машины CentOS, на которой будет установлен сервер Kafka.

Примечание: После установки Apache Kafka рекомендуется создать нового не-root пользователя для работы на сервере.

Как пользователь sudo запустите следующую команду, чтобы создать пользователя kafka:

sudo useradd kafka -m

Флаг -m создаст домашний каталог пользователя. Этот каталог, /home/kafka, будет в дальнейшем использоваться в качестве рабочего пространства для запуска команд.

Установите пароль:

sudo passwd kafka

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

sudo usermod -aG wheel kafka

Пользователь kafka готов к работе. Перейдите в этот пользовательский аккаунт:

su -l kafka

2: Загрузка и извлечение Apache Kafka

Теперь, когда зависимости установлены, можно переходить к загрузке бинарных файлов Apache Kafka.

Создайте каталог Downloads для хранения загруженных пакетов.

mkdir ~/Downloads

Затем используйте curl, чтобы загрузить бинарные файлы Apache Kafka.

curl "http://www-eu.apache.org/dist/kafka/1.1.0/kafka_2.12-1.1.0.tgz" -o ~/Downloads/kafka.tgz

Создайте каталог по имени kafka и откройте его. Это базовый каталог сервера Kafka.

mkdir ~/kafka && cd ~/kafka

Извлеките загруженный архив в этот каталог:

tar -xvzf ~/Downloads/kafka.tgz --strip 1

Флаг —strip 1 извлекает содержимое архива в каталог ~/kafka/, а не в другой подкаталог (например, ~/kafka/kafka_2.12-1.1.0/).

Теперь можно перейти к настройке Kafka.

3: Настройка сервера Kafka

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

vi ~/kafka/config/server.properties

Чтобы изменить стандартное поведение, добавьте в конец файла такую строку:

delete.topic.enable = true

Когда вы закончите, нажмите Esc, чтобы выйти из режима вставки, и :wq, чтобы записать изменения в файл и выйти. Теперь пора перейти к созданию юнит-файлов.

4: Создание юнит-файла для Kafka

Теперь нужно создать юнит-файлы systemd для сервиса Kafka. Это позволит управлять сервисом Kafka – запускать, останавливать и перезапускать его в соответствии с другими сервисами Linux.

Zookeeper – это сервис, который Kafka использует для управления состоянием и конфигурацией кластера. Он широко используется как неотъемлемый компонент во многих распределенных системах. Больше можно узнать в официальной документации Zookeeper.

Создайте файл:

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

В разделе [Unit] указано, что Zookeeper нужна сеть и файловая система, прежде чем сервис сможет начать работу.

Раздел [Service] указывает, что systemd должен использовать файлы оболочки zookeeper-server-start.sh и zookeeper-server-stop.sh  для запуска и остановки сервиса. Он также указывает, что Zookeeper следует перезапускать автоматически, если он выходит из строя.

Теперь создайте юнит-файл для kafka:

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

Раздел [Unit] сообщает, что этот юнит-файл зависит от zookeeper.service. Благодаря этой зависимости zookeeper будет всегда автоматически запускаться раньше, чем kafka.

Раздел [Service] указывает, что systemd должна использовать файлы оболочки kafka-server-start.sh и kafka-server-stop.sh  для запуска и остановки сервиса. Он также указывает, что Kafka следует перезапускать автоматически, если он выходит из строя.

Теперь юнит-файлы готовы. Можно запустить Kafka:

sudo systemctl start kafka

Чтобы убедиться, что сервис запустился успешно, проверьте логи юнита:

journalctl -u kafka

Вы получите:

Jul 17 18:38:59 kafka-ubuntu systemd[1]: Started kafka.service.

Теперь сервис Kafka запущен и прослушивает порт 9092.

Однако пока что этот сервис не будет запускаться автоматически вместе с сервером. Чтобы добавить kafka в автозагрузку, введите:

sudo systemctl enable kafka

5: Тестирование установки

Чтобы убедиться, что сервер Kafka работает правильно, попробуйте опубликовать тестовое сообщение «Hello World».

Для публикации сообщений необходимы:

  • Издатель (producer), который позволяет публиковать записи и данные по темам.
  • Подписчик (consumer), который читает сообщения.

Создайте тему TutorialTopic:

~/kafka/bin/kafka-topics.sh --create --zookeeper localhost:2181 --replication-factor 1 --partitions 1 --topic TutorialTopic

Чтобы создать издателя, используйте в командной строке скрипт kafka-console-producer.sh. Ему нужны аргументы – имя хоста Kafka, порт и тема.

Опубликуйте строку «Hello, World» в теме TutorialTopic:

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

Теперь можно создать подписчика с помощью скрипта kafka-console-consumer.sh. Ему в качестве аргументов нужны имя хоста, порт ZooKeeper и тема.

Следующая команда подпишется на сообщения из TutorialTopic. Обратите внимание на флаг —from-begin, который позволяет читать сообщения, которые были опубликованы до того, как состоялась подписка:

~/kafka/bin/kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic TutorialTopic --from-beginning

Если в конфигурации нет ошибок, вы получите в терминале вывод:

Hello, World

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

Когда вы закончите тестирование, нажмите CTRL + C, чтобы остановить скрипт подписчика.

6: Установка KafkaT (опционально)

KafkaT – это очень полезный инструмент, разработанный Airbnb, который позволяет просматривать подробную информацию о кластере Kafka и выполнять некоторые задачи по управлению из командной строки. Сам инструмент является gem-ом Ruby, потому для его работы нужно установить Ruby. Также понадобится пакет build-essential, позволяющий собрать необходимые gem-ы. Установите всё это при помощи команды:

sudo yum install ruby ruby-devel make gcc patch

Теперь установите KafkaT при помощи команды gem.

sudo gem install kafkat

Конфигурационный файл .kafkatcfg  используется KafkaT для определения каталога установки и логов сервера Kafka. Также он настроит взаимодействие KafkaT и ZooKeeper. Создайте файл:

vi ~/.kafkatcfg

Добавьте в файл следующие строки:

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

Теперь инструмент KafkaT готов к работе. Чтобы просмотреть информацию обо всех разделах сервера Kafka, введите команду:

kafkat partitions

Вывод выглядит так:

Topic                 Partition   Leader      Replicas        ISRs
TutorialTopic         0             0         [0]             [0]
__consumer_offsets    0             0         [0]                           [0]
...
...

Вы увидите TutorialTopic, а также __consumer_offsets, внутреннюю тему, используемую Kafka для хранения информации, связанной с клиентом. Вы можете смело игнорировать строки, начинающиеся с __consumer_offsets.

Примечание: Чтобы узнать больше о KafkaT, посетите GitHub-репозиторий проекта.

7: Настройка многоузлового кластера (опционально)

Чтобы создать многоузловой кластер Kafka, состоящий из нескольких машин CentOS 7, повторите инструкции разделов 1-5 на каждой машине. Кроме того, на каждой машине нужно внести в server.properties такие изменения:

  • Значение broker.id должно быть уникальным на каждом компоненте кластера (например, «server1», «server2» и так далее).
  • Значение параметра zookeeper.connect на каждой ноде должно указывать на один и тот же экземпляр ZooKeeper. Это свойство следует формату <HOSTNAME/IP_ADDRESS>:<PORT> (например, «203.0.113.0:2181», «203.0.113.1:2181»).

Чтобы установить несколько экземпляров ZooKeeper для одного кластера, значение zookeeper.connect на каждой ноде должно содержать IP-адреса и номера портов всех экземпляров ZooKeeper через запятую.

8: Ограничение прав пользователя Kafka

Теперь, когда установка завершена, можно отнять права администратора у пользователя kafka. Но прежде чем сделать это, нужно выйти из системы и снова войти как любой другой не-root пользователь с правами sudo.

Чтобы отнять root-права у пользователя kafka, удалите его из группы wheel:

sudo gpasswd -d kafka wheel

Чтобы повысить безопасность сервера Kafka, заблокируйте пароль пользователя kafka, чтобы никто не мог открыть этот аккаунт.

sudo passwd kafka -l

Теперь только пользователь с правами root или sudo может открыть аккаунт kafka при помощи команды:

sudo su - kafka

Чтобы разблокировать пароль, используйте команду:

sudo passwd kafka -u

Заключение

Теперь сервер Apache Kafka запущен на сервере Ubuntu и защищён. Его можно использовать в различных проектах; клиенты Kafka совместимы с большинством языков программирования. Чтобы узнать больше об Apache Kafka, обратитесь к документации.

Tags: , , , , ,