Apache Kafka: основные операции
Предлагаю сначала реализовать конфигурацию «один узел - один брокер», а затем - конфигурацию «один узел - несколько брокеров».
Что у Вас уже должно быть установлено: Java, ZooKeeper и Kafka.
Прежде, чем мы перейдем к настройке кластера Kafka, необходимо запустить ZooKeeper, поскольку кластер Kafka использует ZooKeeper.
Запускаем ZooKeeper
Открываем терминал и вводим следующую команду:
bin/zookeeper-server-start.sh config/zookeeper.properties
Запускаем брокер Kafka
bin/kafka-server-start.sh config/server.properties
После запуска Kafka Broker введите команду jps на терминале ZooKeeper, Вы увидите следующее сообщение:
821 QuorumPeerMain 928 Kafka 931 Jps
Кроме того, Вы увидите два демона, запущенных на терминале, из которых QuorumPeerMain — это демон ZooKeeper, а другой — демон Kafka.
Конфигурация «один узел - один брокер»
Данная конфигурация подразумевает наличие экземпляра ZooKeeper и id брокера. Алгоритм настройки:
1) Создание топика Kafka: Для создания тем на сервере Kafka используйте утилиту командной строки под названием kafka-topics.sh. Откройте новый терминал и введите пример, приведенный ниже.
2) Синтаксис
bin/kafka-topics.sh --create --zookeeper localhost:2181 --replication-factor 1 --partitions 1 --topic topic-name
Пример
bin/kafka-topics.sh --create --zookeeper localhost:2181 --replication-factor 1 --partitions 1 --topic Hello-Kafka
Таким образом, мы создали топик под названием Hello-Kafka с одним разделом и одним фактором реплики. Вывод, созданный выше, будет примерно следующий:
3) Вывод – Создан топик «Hello-Kafka»
После создания топика Вы в окне терминала брокера Kafka Вы увидите уведомление об успешном завершении операции. Кроме того, для созданного топика в папке «tmp/kafka-logs/» в «/в файле config/server.properties появится специальный журнал.
Список топиков
Для того, чтобы получить список топиков на сервере Kafka, введите следующую команду:
Синтаксис
bin/kafka-topics.sh --list --zookeeper localhost:2181
Вывод
Hello-Kafka
Поскольку мы создали только 1 топик, в результате выполнения команды мы увидим только 1 название – «Hello-Kafka». Если бы мы создали несколько топик, то на выходе мы бы получили несколько названий созданных нами топиков.
Запуск продюсера для отправки сообщений
Синтаксис
bin/kafka-console-producer.sh --broker-list localhost:9092 --topic topic-name
Из приведенного выше синтаксиса следует, что для клиента командной строки производителя требуются два основных параметра −
Broker-list – это список брокеров, которым мы хотим отправлять сообщения. В нашем случае у нас есть только один брокер. Файл Config/server.properties содержит идентификатор порта брокера. Посколько мы знаем, что наш брокер слушает порт 9092, мы можем указать его напрямую.
Kafka Topic – пример топика в Apache Kafka.
Пример
bin/kafka-console-producer.sh --broker-list localhost:9092 --topic Hello-Kafka
Продюсер будет ожидать ввода данных из stdin и опубликует их в кластере Kafka. По умолчанию каждая новая строка будет опубликована как новое сообщение, свойства производителя по умолчанию задаются в файле config/producer.properties. Теперь наберите несколько строк сообщений в терминале, как показано ниже.
Вывод
$ bin/kafka-console-producer.sh --broker-list localhost:9092 --topic Hello-Kafka[2016-01-16 13:50:45,931] WARN property topic is not valid (kafka.utils.Verifia-bleProperties) Hello My first message
Запустите консьюмера, получающего сообщения
Как и в случае с продюсером, свойства потребителя задаются в файле config/consumer.proper-ties. Откройте новый терминал и введите следующий синтаксис:
Синтаксис
bin/kafka-console-consumer.sh --zookeeper localhost:2181 —topic topic-name --from-beginning
Пример
bin/kafka-console-consumer.sh --zookeeper localhost:2181 —topic Hello-Kafka --from-beginning
Вывод
Hello My first message My second message
Теперь Вы можете ввести сообщения из терминала продюсера после чего они появятся в терминале консьюмера.
Итак, мы поговорили о кластере с одним узлом и одним брокером. Теперь давайте перейдем к конфигурации с несколькими брокерами.
Конфигурация «один узел - несколько брокеров»
Перед настройкой кластера с несколькими брокерами запустим сервер ZooKeeper.
Создание нескольких брокеров Kafka - у нас еже есть 1 созданный экземпляр, но нам необходимо несколько экземпляров. Поэтому мы копируем существующие настройки сервера в 2 новых конфигурационных файла и присваиваем им имя server-one.properties и server-two.properties соответственно. Затем редактируем эти два файла и вносим в них следующие изменения:
config/server-one.properties
# The id of the broker. This m ust be set to a unique integer for each broker. broker.id=1 # The port the socket server listens on port=9093 # A comma seperated list of directories under which to store log files log.dirs=/tmp/kafka-logs-1
config/server-two.properties
# The id of the broker. This must be set to a unique integer for each broker. broker.id=2 # The port the socket server listens on port=9094 # A comma seperated list of directories under which to store log files log.dirs=/tmp/kafka-logs-2
Запуск нескольких брокеров − После внесения всех изменений на трех серверах откройте три новых терминала для запуска каждого брокера по очереди.
Broker1
bin/kafka-server-start.sh config/server.properties
Broker2
bin/kafka-server-start.sh config/server-one.properties
Broker3
bin/kafka-server-start.sh config/server-two.properties
Теперь у нас есть три разных брокера. Попробуйте последовательно запустить их.
Создание топика
Поскольку у нас работают три разных брокера, назначим значение коэффициента репликации равным трем. Если у Вас работают два брокера, значение коэффициента репликации должно быть равно двум.
Синтаксис
bin/kafka-topics.sh --create --zookeeper localhost:2181 –replica tion-factor 3 -partitions 1 --topic topic-name
Пример
bin/kafka-topics.sh --create --zookeeper localhost:2181 --replication-factor 3 -partitions 1 --topic Multibrokerapplication
Вывод
Создание топика “Multibrokerapplication”
Команда Describe используется для проверки того, какой брокер прослушивает текущий топик, как показано ниже:
bin/kafka-topics.sh --describe --zookeeper localhost:2181 --topic Multibrokerappli-cation
Вывод
bin/kafka-topics.sh --describe --zookeeper localhost:2181 --topic Multibrokerappli-cation Topic:Multibrokerapplication PartitionCount:1 ReplicationFactor:3 Configs: Topic:Multibrokerapplication Partition:0 Leader:0 Replicas:0,2,1 Isr:0,2,1
Таким образом, можно сделать вывод, что первая строка дает сводку по всем разделам, показывая название топика, количество разделов и коэффициент репликации, который мы выбрали.
В нашем случае мы видим, что наш первый брокер (с broker.id 0) является лидером. Тогда Replicas:0,2,1 означает, что все брокеры реплицируют эту тему, Isr - это набор синхронизированных реплик. Это подмножество реплик, которые в данный момент “живы” и управляются лидером.
Запуск продюсера для отправки сообщений
Процедура такая же, как и в случае с одним брокером.
Пример
bin/kafka-console-producer.sh --broker-list localhost:9092 --topic Multibrokerapplication
Вывод
bin/kafka-console-producer.sh --broker-list localhost:9092 --topic Multibrokerapplication [2016-01-20 19:27:21,045] WARN Property topic is not valid (kafka.utils.Verifia-bleProperties) This is single node-multi broker demo This is the second message
Запустите консьюмера, получающего сообщения
Процедура такая же, как и в случае с одним брокером
Пример
bin/kafka-console-consumer.sh --zookeeper localhost:2181 —topic Multibrokerapplica-tion --from-beginning
Вывод
bin/kafka-console-consumer.sh --zookeeper localhost:2181 —topic Multibrokerapplica-tion —from-beginning This is single node-multi broker demo This is the second message
Основные операции с новыми топиками
В этой главе мы рассмотрим различные базовые операции с топиками.
Изменение топика
Вы уже поняли, как создать топик в кластере Kafka. Теперь давайте изменим созданную тему с помощью следующей команды:
Синтаксис:
bin/kafka-topics.sh —zookeeper localhost:2181 --alter --topic topic_name --parti-tions count
Пример
bin/kafka-topics.sh --zookeeper localhost:2181 --alter --topic Hello-kafka --parti-tions 2
Вывод
ВНИМАНИЕ: Увеличение числа разделов для топиков, содержащих ключи, приведете к упорядочиванию сообщений. Добавление разделов прошло успешно.
Удаление топика
Для удаления топика используйте следующий синтаксис:
Синтаксис
bin/kafka-topics.sh --zookeeper localhost:2181 --delete --topic topic_name
Пример
bin/kafka-topics.sh --zookeeper localhost:2181 --delete --topic Hello-kafka
Вывод
> Topic Hello-kafka marked for deletion
Примечание: если для параметра delete.topic.enable не указано значение true, данная операция не будет обработана.
