Быстрый старт с Kafka

февраля 06, 2021

Шаг 1: Получить Kafka

Загрузите последнюю версию Kafka и распакуйте ее:

$ tar -xzf kafka_2.13-2.7.0.tgz
$ cd kafka_2.13-2.7.0

Шаг 2: Запустить среду Kafka

Примечание. В вашей локальной среде должна быть установлена Java 8+.

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

# Запустить сервис ZooKeeper
# Примечание. 
# Скоро ZooKeeper больше не будет требоваться для Apache Kafka.
$ bin/zookeeper-server-start.sh config/zookeeper.properties

Откройте другой сеанс терминала и запустите:

# Запустить брокерскую службу Kafka
$ bin/kafka-server-start.sh config/server.properties

После успешного запуска всех служб у вас будет запущена и готова к использованию базовая среда Kafka.

Шаг 3: Создать тему (topic) для сохранения событий

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

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

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

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

Все инструменты командной строки Kafka имеют дополнительные параметры: запустите команду kafka-topics.sh без каких-либо аргументов, чтобы отобразить информацию об использовании. Например, он также может показать вам такие подробности, как количество разделов новой темы:

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

Topic:quickstart-events  PartitionCount:1    ReplicationFactor:1 Configs:
    Topic: quickstart-events Partition: 0    Leader: 0   Replicas: 0 Isr: 0

Шаг 4: Запишите события в тему

Клиент Kafka связывается с брокерами Kafka через сеть для записи (или чтения) событий. После получения брокеры будут хранить события в надежном и отказоустойчивом виде столько времени, сколько вам нужно - даже навсегда.

Запустите консольный клиент производителя (producer client), чтобы записать несколько событий в вашу тему. По умолчанию каждая вводимая вами строка приводит к записи отдельного события в тему.

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

This is my first event
This is my second event

Вы можете остановить клиента производителя с помощью Ctrl-C в любое время.

Шаг 5: Прочитать события

Откройте другой сеанс терминала и запустите клиентский консольный клиент, чтобы прочитать только что созданные события:

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

This is my first event
This is my second event

Вы можете остановить клиентского клиента с помощью Ctrl-C в любое время.

Не стесняйтесь экспериментировать: например, переключитесь обратно на свой терминал производителя (предыдущий шаг), чтобы записать дополнительные события, и посмотрите, как события немедленно отображаются в вашем потребительском терминале.

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

Шаг 6: Импорт/экспорт данных как поток событий с Kafka Connect

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

Шаг 7: Обработка событий с Kafka Streams

Как только ваши данные будут сохранены в Kafka как события, вы можете обрабатывать их с помощью клиентской библиотеки Kafka Streams для Java/Scala. Она позволяет вам реализовать критически важные приложения и микросервисы реального времени, где входные и/или выходные данные хранятся в темах Kafka. Kafka Streams сочетает в себе простоту написания и развертывания стандартных приложений Java и Scala на стороне клиента с преимуществами серверной кластерной технологии Kafka, чтобы сделать эти приложения высокомасштабируемыми, эластичными, отказоустойчивыми и распределенными. Библиотека поддерживает однократную обработку, операции и агрегирование с отслеживанием состояния, управление окнами, объединения, обработку на основе времени события и многое другое.

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

KStream<String, String> textLines = builder.stream("quickstart-events");

KTable<String, Long> wordCounts = textLines
            .flatMapValues(line -> Arrays.asList(line.toLowerCase().split(" ")))
            .groupBy((keyIgnored, word) -> word)
            .count();

wordCounts.toStream().to("output-topic"), Produced.with(Serdes.String(), Serdes.Long()));

Шаг 8: Свернуть среду Kafka

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

Остановите клиентов-производителей и потребителей с помощью Ctrl-C, если вы еще этого не сделали.
Остановите брокера Kafka с помощью Ctrl-C.
Наконец, остановите сервер ZooKeeper с помощью Ctrl-C.

Если вы также хотите удалить какие-либо данные вашей локальной среды Kafka, включая любые события, созданные вами в процессе, выполните команду:

$ rm -rf /tmp/kafka-logs /tmp/zookeeper

Поиск по этому блогу

Alex tools