Apache Kafka - Краткое руководство
В больших данных используется огромный объем данных. Что касается данных, у нас есть две основные проблемы: первая задача - как собрать большой объем данных, а вторая - проанализировать собранные данные. Чтобы преодолеть эти проблемы, вам понадобится система обмена сообщениями.
Kafka разработан для распределенных систем с высокой пропускной способностью. Kafka, как правило, очень хорошо работает как замена более традиционному брокеру сообщений. По сравнению с другими системами обмена сообщениями, Kafka имеет лучшую пропускную способность, встроенное разделение, репликацию и присущую отказоустойчивость, что делает его подходящим для крупномасштабных приложений обработки сообщений.
Что такое система обмена сообщениями?
Система обмена сообщениями отвечает за передачу данных из одного приложения в другое, поэтому приложения могут сосредоточиться на данных, но не беспокоиться о том, как ими поделиться. Распределенный обмен сообщениями основан на концепции надежной организации очереди сообщений. Сообщения ставятся в очередь асинхронно между клиентскими приложениями и системой обмена сообщениями. Доступны два типа шаблонов обмена сообщениями: один - точка-точка, а другой - система обмена сообщениями публикация-подписка (pub-sub). Большинство шаблонов обмена сообщениями следуютpub-sub.
Система обмена сообщениями точка-точка
В системе точка-точка сообщения сохраняются в очереди. Один или несколько потребителей могут использовать сообщения в очереди, но конкретное сообщение может быть использовано максимум одним потребителем. Как только потребитель прочитает сообщение в очереди, оно исчезает из этой очереди. Типичным примером этой системы является система обработки заказов, в которой каждый заказ будет обрабатываться одним обработчиком заказов, но одновременно могут работать и несколько обработчиков заказов. На следующей диаграмме изображена структура.
Система обмена сообщениями публикации и подписки
В системе публикации-подписки сообщения сохраняются в теме. В отличие от двухточечной системы, потребители могут подписаться на одну или несколько тем и использовать все сообщения в этой теме. В системе публикации-подписки производители сообщений называются издателями, а потребители сообщений - подписчиками. Примером из реальной жизни является Dish TV, на котором публикуются различные каналы, такие как спорт, фильмы, музыка и т. Д., И каждый может подписаться на свой собственный набор каналов и получать их, когда доступны каналы, на которые они подписаны.
Что такое Кафка?
Apache Kafka - это распределенная система обмена сообщениями "публикация-подписка" и надежная очередь, которая может обрабатывать большой объем данных и позволяет передавать сообщения от одной конечной точки к другой. Kafka подходит как для автономного, так и для онлайн-потребления сообщений. Сообщения Kafka сохраняются на диске и реплицируются в кластере для предотвращения потери данных. Kafka построен на основе службы синхронизации ZooKeeper. Он очень хорошо интегрируется с Apache Storm и Spark для анализа потоковых данных в реальном времени.
Льготы
Ниже приведены несколько преимуществ Kafka:
Reliability - Kafka распределен, разделен, реплицирован и отказоустойчив.
Scalability - Система обмена сообщениями Kafka легко масштабируется без простоев.
Durability- Kafka использует
журнал распределенных фиксаций,
что означает, что сообщения сохраняются на диске как можно быстрее, следовательно, они надежны.Performance- Kafka имеет высокую пропускную способность как для публикации, так и для подписки на сообщения. Он поддерживает стабильную производительность, даже если хранится много ТБ сообщений.
Kafka работает очень быстро и гарантирует нулевое время простоя и нулевую потерю данных.
Случаи использования
Kafka можно использовать во многих случаях использования. Некоторые из них перечислены ниже -
Metrics- Кафка часто используется для оперативного мониторинга данных. Это включает агрегирование статистики из распределенных приложений для создания централизованных потоков операционных данных.
Log Aggregation Solution - Kafka можно использовать в рамках всей организации для сбора журналов из нескольких служб и предоставления их в стандартном формате нескольким потребителям.
Stream Processing- Популярные платформы, такие как Storm и Spark Streaming, считывают данные из темы, обрабатывают их и записывают обработанные данные в новую тему, где они становятся доступными для пользователей и приложений. Высокая надежность Kafka также очень полезна в контексте потоковой обработки.
Потребность в Кафке
Kafka - это единая платформа для обработки всех потоков данных в реальном времени. Kafka поддерживает доставку сообщений с малой задержкой и гарантирует отказоустойчивость при сбоях машины. Он способен обслуживать большое количество разнообразных потребителей. Kafka работает очень быстро, выполняет 2 миллиона операций записи в секунду. Kafka сохраняет все данные на диск, что по сути означает, что все записи идут в кеш страниц ОС (ОЗУ). Это делает очень эффективным перенос данных из кеша страниц в сетевой сокет.
Прежде чем углубляться в Kafka, вы должны знать основные термины, такие как темы, брокеры, производители и потребители. На следующей диаграмме показаны основные термины, а в таблице подробно описаны компоненты диаграммы.
На приведенной выше диаграмме тема разделена на три раздела. Раздел 1 имеет два фактора смещения 0 и 1. Раздел 2 имеет четыре фактора смещения 0, 1, 2 и 3. Раздел 3 имеет один коэффициент смещения 0. Идентификатор реплики совпадает с идентификатором сервера, на котором она размещена.
Предположим, что если коэффициент репликации темы установлен на 3, то Kafka создаст 3 идентичные реплики каждого раздела и поместит их в кластер, чтобы сделать их доступными для всех своих операций. Чтобы сбалансировать нагрузку в кластере, каждый брокер хранит один или несколько таких разделов. Несколько производителей и потребителей могут публиковать и получать сообщения одновременно.
S.No | Компоненты и описание |
---|---|
1 | Topics Поток сообщений, принадлежащих к определенной категории, называется темой. Данные хранятся в темах. Темы разбиты на разделы. Для каждой темы у Kafka есть минимум один раздел. Каждый такой раздел содержит сообщения в неизменной упорядоченной последовательности. Раздел реализован как набор файлов сегментов одинакового размера. |
2 | Partition Темы могут иметь много разделов, поэтому они могут обрабатывать произвольный объем данных. |
3 | Partition offset Каждое разделенное на разделы сообщение имеет уникальный идентификатор последовательности, называемый |
4 | Replicas of partition Реплики - это не что иное, как |
5 | Brokers
|
6 | Kafka Cluster Kafka имеет более одного брокера, что называется кластером Kafka. Кластер Kafka можно расширить без простоев. Эти кластеры используются для управления сохранением и репликацией данных сообщений. |
7 | Producers Продюсеры - это издатели сообщений по одной или нескольким темам Kafka. Производители отправляют данные брокерам Kafka. Каждый раз, когда производитель публикует сообщение брокеру, брокер просто добавляет сообщение в последний файл сегмента. Фактически, сообщение будет добавлено к разделу. Производитель также может отправлять сообщения в раздел по своему выбору. |
8 | Consumers Потребители читают данные от брокеров. Потребители подписываются на одну или несколько тем и используют опубликованные сообщения, получая данные от брокеров. |
9 | Leader
|
10 | Follower Узел, который следует инструкциям лидера, называется подчиненным. Если лидер терпит неудачу, один из последователей автоматически становится новым лидером. Последователь действует как обычный потребитель, извлекает сообщения и обновляет собственное хранилище данных. |
Взгляните на следующую иллюстрацию. Он показывает кластерную диаграмму Kafka.
В следующей таблице описан каждый из компонентов, показанных на диаграмме выше.
S.No | Компоненты и описание |
---|---|
1 | Broker Кластер Kafka обычно состоит из нескольких брокеров для поддержания баланса нагрузки. Брокеры Kafka не имеют состояния, поэтому они используют ZooKeeper для поддержания состояния своего кластера. Один экземпляр брокера Kafka может обрабатывать сотни тысяч операций чтения и записи в секунду, и каждый брокер может обрабатывать ТБ сообщений без снижения производительности. Выбор лидера брокера Kafka может быть произведен ZooKeeper. |
2 | ZooKeeper ZooKeeper используется для управления и координации брокера Kafka. Служба ZooKeeper в основном используется для уведомления производителя и потребителя о присутствии любого нового брокера в системе Kafka или сбое брокера в системе Kafka. В соответствии с полученным Zookeeper уведомлением о присутствии или отказе брокера производитель и потребитель принимают решение и начинают согласовывать свою задачу с другим брокером. |
3 | Producers Производители отправляют данные брокерам. Когда новый брокер запускается, все производители ищут его и автоматически отправляют сообщение этому новому брокеру. Производитель Kafka не ждет подтверждений от брокера и отправляет сообщения настолько быстро, насколько брокер может их обработать. |
4 | Consumers Поскольку брокеры Kafka не имеют состояния, это означает, что потребитель должен поддерживать, сколько сообщений было использовано, используя смещение раздела. Если потребитель подтверждает конкретное смещение сообщения, это означает, что он использовал все предыдущие сообщения. Потребитель отправляет брокеру асинхронный пул-реквест, чтобы получить буфер байтов, готовый к использованию. Потребители могут перемотать назад или перейти к любой точке раздела, просто указав значение смещения. ZooKeeper сообщает значение смещения потребителя. |
На данный момент мы обсудили основные концепции Kafka. Давайте теперь проливаем свет на рабочий процесс Kafka.
Kafka - это просто набор тем, разбитых на один или несколько разделов. Раздел Kafka - это линейно упорядоченная последовательность сообщений, где каждое сообщение идентифицируется своим индексом (называемым смещением). Все данные в кластере Kafka представляют собой несвязанное объединение разделов. Входящие сообщения записываются в конце раздела, и сообщения последовательно читаются потребителями. Долговечность обеспечивается репликацией сообщений разным брокерам.
Kafka предоставляет систему обмена сообщениями на основе pub-sub и очередей в быстром, надежном, устойчивом, отказоустойчивом и нулевом времени простоя. В обоих случаях производители просто отправляют сообщение в тему, а потребитель может выбрать любой тип системы обмена сообщениями в зависимости от своих потребностей. Давайте проследим шаги в следующем разделе, чтобы понять, как потребитель может выбрать систему обмена сообщениями по своему выбору.
Рабочий процесс обмена сообщениями Pub-Sub
Ниже приведен пошаговый рабочий процесс обмена сообщениями Pub-Sub.
Производители регулярно отправляют сообщения в тему.
Брокер Kafka хранит все сообщения в разделах, настроенных для этой конкретной темы. Это гарантирует, что сообщения равномерно распределяются между разделами. Если производитель отправляет два сообщения и есть два раздела, Kafka сохранит одно сообщение в первом разделе, а второе сообщение во втором разделе.
Потребитель подписывается на определенную тему.
После того, как потребитель подписывается на тему, Kafka предоставит текущее смещение темы потребителю, а также сохранит смещение в ансамбле Zookeeper.
Потребитель будет запрашивать у Kafka регулярные интервалы (например, 100 мс) для получения новых сообщений.
Как только Kafka получает сообщения от производителей, он пересылает эти сообщения потребителям.
Потребитель получит сообщение и обработает его.
Как только сообщения будут обработаны, потребитель отправит подтверждение брокеру Kafka.
Как только Kafka получает подтверждение, он изменяет смещение на новое значение и обновляет его в Zookeeper. Поскольку смещения поддерживаются в Zookeeper, потребитель может правильно прочитать следующее сообщение даже во время сбоев сервера.
Этот процесс будет повторяться до тех пор, пока потребитель не остановит запрос.
Потребитель может в любой момент перемотать назад / перейти к желаемому смещению темы и прочитать все последующие сообщения.
Рабочий процесс обмена сообщениями в очереди / Группа потребителей
В системе обмена сообщениями очереди вместо одного потребителя группа потребителей с одинаковым идентификатором группы
будет подписываться на тему. Проще говоря, потребители, подписывающиеся на тему с одинаковым идентификатором группы
, считаются единой группой, и сообщения распределяются между ними. Давайте проверим реальный рабочий процесс этой системы.
Продюсеры регулярно отправляют сообщения в тему.
Kafka хранит все сообщения в разделах, настроенных для этой конкретной темы, как и в предыдущем сценарии.
Один потребитель подписывается на определенную тему, предположим, что
Тема-01
сидентификатором
группы
какГруппа-1
.Kafka взаимодействует с потребителем так же, как и обмен сообщениями Pub-Sub, до тех пор, пока новый потребитель не подписывается на ту же тему,
Тема-01,
с тем жеидентификатором
группы,
что игруппа-1
.Как только появляется новый потребитель, Kafka переключает свою работу в режим совместного использования и делится данными между двумя потребителями. Это совместное использование будет продолжаться до тех пор, пока количество потребителей не достигнет количества разделов, настроенных для этой конкретной темы.
Как только количество потребителей превысит количество разделов, новый потребитель не получит никаких дальнейших сообщений, пока любой из существующих потребителей не отменит подписку. Этот сценарий возникает из-за того, что каждому потребителю в Kafka будет назначен минимум один раздел, и как только все разделы будут назначены существующим потребителям, новым потребителям придется ждать.
Эта функция также называется
группой потребителей
. Таким же образом Kafka предоставит лучшее из обеих систем очень простым и эффективным способом.
Роль ZooKeeper
Важной зависимостью Apache Kafka является Apache Zookeeper, который представляет собой распределенную службу конфигурации и синхронизации. Zookeeper служит координационным интерфейсом между брокерами Kafka и потребителями. Серверы Kafka обмениваются информацией через кластер Zookeeper. Kafka хранит в Zookeeper основные метаданные, такие как информация о темах, брокерах, смещениях потребителей (считыватели очереди) и так далее.
Поскольку вся критическая информация хранится в Zookeeper, и он обычно реплицирует эти данные по всему ансамблю, отказ брокера Kafka / Zookeeper не влияет на состояние кластера Kafka. Kafka восстановит состояние после перезапуска Zookeeper. Это дает нулевое время простоя для Kafka. Выбор лидера между брокером Kafka также осуществляется с помощью Zookeeper в случае отказа лидера.
Чтобы узнать больше о Zookeeper, обратитесь к zookeeper
Давайте продолжим дальше, как установить Java, ZooKeeper и Kafka на ваш компьютер, в следующей главе.
Ниже приведены шаги по установке Java на ваш компьютер.
Шаг 1. Проверка установки Java
Надеюсь, вы уже установили java на свой компьютер прямо сейчас, поэтому просто проверьте его, используя следующую команду.
$ java -version
Если Java успешно установлена на вашем компьютере, вы можете увидеть версию установленной Java.
Шаг 1.1 - Загрузите JDK
Если Java не загружена, загрузите последнюю версию JDK, перейдя по следующей ссылке, и загрузите последнюю версию.
http://www.oracle.com/technetwork/java/javase/downloads/index.htmlТеперь последняя версия - JDK 8u 60, а файл - «jdk-8u60-linux-x64.tar.gz». Загрузите файл на свой компьютер.
Шаг 1.2 - Извлечение файлов
Обычно загружаемые файлы хранятся в папке загрузок, проверьте ее и извлеките установку tar с помощью следующих команд.
$ cd /go/to/download/path
$ tar -zxf jdk-8u60-linux-x64.gz
Шаг 1.3 - Перейти в каталог Opt
Чтобы сделать Java доступной для всех пользователей, переместите извлеченное содержимое Java в
папку usr / local / java
/.
$ su
password: (type password of root user)
$ mkdir /opt/jdk $ mv jdk-1.8.0_60 /opt/jdk/
Шаг 1.4 - Установите путь
Чтобы установить путь и переменные JAVA_HOME, добавьте следующие команды в файл ~ / .bashrc.
export JAVA_HOME =/usr/jdk/jdk-1.8.0_60
export PATH=$PATH:$JAVA_HOME/bin
Теперь примените все изменения к текущей работающей системе.
$ source ~/.bashrc
Шаг 1.5 - Альтернативы Java
Используйте следующую команду, чтобы изменить альтернативы Java.
update-alternatives --install /usr/bin/java java /opt/jdk/jdk1.8.0_60/bin/java 100
Step 1.6 - Теперь проверьте Java с помощью команды проверки (java -version), описанной в шаге 1.
Шаг 2 - Установка ZooKeeper Framework
Шаг 2.1 - Загрузите ZooKeeper
Чтобы установить ZooKeeper framework на свой компьютер, перейдите по следующей ссылке и загрузите последнюю версию ZooKeeper.
http://zookeeper.apache.org/releases.htmlНа данный момент последняя версия ZooKeeper - 3.4.6 (ZooKeeper-3.4.6.tar.gz).
Шаг 2.2 - Извлеките tar-файл
Извлеките tar-файл, используя следующую команду
$ cd opt/
$ tar -zxf zookeeper-3.4.6.tar.gz $ cd zookeeper-3.4.6
$ mkdir data
Шаг 2.3 - Создайте файл конфигурации
Откройте файл конфигурации с именем conf / zoo.cfg,
используя команду vi «conf / zoo.cfg» и все следующие параметры, чтобы установить их в качестве отправной точки.
$ vi conf/zoo.cfg
tickTime=2000
dataDir=/path/to/zookeeper/data
clientPort=2181
initLimit=5
syncLimit=2
После того, как файл конфигурации будет успешно сохранен и снова вернется в терминал, вы можете запустить сервер zookeeper.
Шаг 2.4 - Запустите сервер ZooKeeper
$ bin/zkServer.sh start
После выполнения этой команды вы получите ответ, как показано ниже -
$ JMX enabled by default
$ Using config: /Users/../zookeeper-3.4.6/bin/../conf/zoo.cfg $ Starting zookeeper ... STARTED
Шаг 2.5 - Запустите CLI
$ bin/zkCli.sh
После ввода указанной выше команды вы будете подключены к серверу zookeeper и получите ответ, указанный ниже.
Connecting to localhost:2181
................
................
................
Welcome to ZooKeeper!
................
................
WATCHER::
WatchedEvent state:SyncConnected type: None path:null
[zk: localhost:2181(CONNECTED) 0]
Шаг 2.6 - Остановка сервера Zookeeper
После подключения сервера и выполнения всех операций вы можете остановить сервер zookeeper с помощью следующей команды -
$ bin/zkServer.sh stop
Теперь вы успешно установили Java и ZooKeeper на свой компьютер. Давайте посмотрим, как установить Apache Kafka.
Шаг 3 - установка Apache Kafka
Давайте продолжим следующие шаги, чтобы установить Kafka на ваш компьютер.
Шаг 3.1 - Загрузите Kafka
Чтобы установить Kafka на свой компьютер, щелкните ссылку ниже -
https://www.apache.org/dyn/closer.cgi?path=/kafka/0.9.0.0/kafka_2.11-0.9.0.0.tgzТеперь последняя версия, то есть - kafka_2.11_0.9.0.0.tgz будет загружен на ваш компьютер.
Шаг 3.2 - Извлеките tar-файл
Извлеките файл tar, используя следующую команду -
$ cd opt/ $ tar -zxf kafka_2.11.0.9.0.0 tar.gz
$ cd kafka_2.11.0.9.0.0
Теперь вы загрузили последнюю версию Kafka на свой компьютер.
Шаг 3.3 - Запуск сервера
Вы можете запустить сервер, введя следующую команду -
$ bin/kafka-server-start.sh config/server.properties
После запуска сервера вы увидите на экране следующий ответ:
$ bin/kafka-server-start.sh config/server.properties
[2016-01-02 15:37:30,410] INFO KafkaConfig values:
request.timeout.ms = 30000
log.roll.hours = 168
inter.broker.protocol.version = 0.9.0.X
log.preallocate = false
security.inter.broker.protocol = PLAINTEXT
…………………………………………….
…………………………………………….
Шаг 4 - Остановите сервер
После выполнения всех операций вы можете остановить сервер с помощью следующей команды -
$ bin/kafka-server-stop.sh config/server.properties
Теперь, когда мы уже обсудили установку Kafka, мы можем узнать, как выполнять основные операции с Kafka в следующей главе.
Сначала давайте приступим к реализации конфигурации с одним узлом и одним брокером,
а затем перенесем нашу установку на конфигурацию с одним узлом и несколькими брокерами.
Надеюсь, вы уже установили Java, ZooKeeper и Kafka на свой компьютер. Перед тем как перейти к настройке кластера Kafka, сначала вам нужно будет запустить ZooKeeper, потому что кластер Kafka использует ZooKeeper.
Запустить ZooKeeper
Откройте новый терминал и введите следующую команду -
bin/zookeeper-server-start.sh config/zookeeper.properties
Чтобы запустить Kafka Broker, введите следующую команду -
bin/kafka-server-start.sh config/server.properties
После запуска Kafka Broker введите команду jps
в терминале ZooKeeper, и вы увидите следующий ответ:
821 QuorumPeerMain
928 Kafka
931 Jps
Теперь вы могли видеть два демона, работающих на терминале, где QuorumPeerMain - это демон ZooKeeper, а другой - демон Kafka.
Конфигурация с одним узлом и одним брокером
В этой конфигурации у вас есть один экземпляр ZooKeeper и идентификатор брокера. Ниже приведены шаги по его настройке -
Creating a Kafka Topic- Kafka предоставляет утилиту командной строки с именем kafka-topics.sh
для создания тем на сервере. Откройте новый терминал и введите приведенный ниже пример.
Syntax
bin/kafka-topics.sh --create --zookeeper localhost:2181 --replication-factor 1
--partitions 1 --topic topic-name
Example
bin/kafka-topics.sh --create --zookeeper localhost:2181 --replication-factor 1
--partitions 1 --topic Hello-Kafka
Мы только что создали тему Hello-Kafka
с одним разделом и одним фактором реплики. Созданный выше вывод будет похож на следующий вывод -
Output- Создал тему Hello-Kafka
После создания темы вы можете получить уведомление в окне терминала брокера Kafka и журнал для созданной темы, указанный в «/ tmp / kafka-logs /» в файле config / server.properties.
Список тем
Чтобы получить список тем на сервере Kafka, вы можете использовать следующую команду -
Syntax
bin/kafka-topics.sh --list --zookeeper localhost:2181
Output
Hello-Kafka
Поскольку мы создали тему, в ней будет отображаться
только Hello-Kafka
. Предположим, что если вы создаете более одной темы, вы получите названия тем на выходе.
Запустите Producer для отправки сообщений
Syntax
bin/kafka-console-producer.sh --broker-list localhost:9092 --topic topic-name
Из приведенного выше синтаксиса для клиента командной строки производителя требуются два основных параметра:
Broker-list- Список брокеров, которым мы хотим отправлять сообщения. В этом случае у нас только один брокер. Файл Config / server.properties содержит идентификатор порта брокера, поскольку мы знаем, что наш брокер прослушивает порт 9092, поэтому вы можете указать его напрямую.
Название темы - вот пример названия темы.
Example
bin/kafka-console-producer.sh --broker-list localhost:9092 --topic Hello-Kafka
Производитель будет ждать ввода от стандартного ввода и опубликовать в кластере Kafka. По умолчанию каждая новая строка публикуется как новое сообщение, тогда свойства производителя по умолчанию указываются в файле config /
Producer.properties. Теперь вы можете ввести несколько строк сообщений в терминал, как показано ниже.
Output
$ 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
My second message
Запустить потребителя для получения сообщений
Как и в случае с производителем, потребительские свойства по умолчанию указаны в файле config / consumer.proper -ties
. Откройте новый терминал и введите приведенный ниже синтаксис для получения сообщений.
Syntax
bin/kafka-console-consumer.sh --zookeeper localhost:2181 —topic topic-name
--from-beginning
Example
bin/kafka-console-consumer.sh --zookeeper localhost:2181 —topic Hello-Kafka
--from-beginning
Output
Hello
My first message
My second message
Наконец, вы можете вводить сообщения с терминала производителя и видеть, как они появляются в терминале потребителя. На данный момент у вас есть очень хорошее представление о кластере с одним узлом и одним брокером. Теперь перейдем к настройке нескольких брокеров.
Конфигурация одного узла и нескольких брокеров
Прежде чем перейти к настройке кластера с несколькими брокерами, сначала запустите сервер ZooKeeper.
Create Multiple Kafka Brokers- У нас уже есть один экземпляр брокера Kafka в файле con-fig / server.properties. Теперь нам нужно несколько экземпляров брокера, поэтому скопируйте существующий файл server.prop-erties в два новых файла конфигурации и переименуйте его как server-one.properties и server-two.prop-erties. Затем отредактируйте оба новых файла и назначьте следующие изменения -
config / server-one.properties
# The id of the broker. This must 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
Start Multiple Brokers- После внесения всех изменений на трех серверах откройте три новых терминала, чтобы запускать каждого брокера по очереди.
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
Теперь у нас есть три разных брокера, работающих на машине. Попробуйте сами проверить всех демонов, набравjps на терминале ZooKeeper, то вы увидите ответ.
Создание темы
Давайте назначим коэффициент репликации равным трем для этой темы, потому что у нас работает три разных брокера. Если у вас два брокера, то назначенное значение реплики будет равно двум.
Syntax
bin/kafka-topics.sh --create --zookeeper localhost:2181 --replication-factor 3
-partitions 1 --topic topic-name
Example
bin/kafka-topics.sh --create --zookeeper localhost:2181 --replication-factor 3
-partitions 1 --topic Multibrokerapplication
Output
created topic “Multibrokerapplication”
Команда Describe
используется для проверки того, какой брокер прослушивает текущую созданную тему, как показано ниже -
bin/kafka-topics.sh --describe --zookeeper localhost:2181
--topic Multibrokerappli-cation
Output
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
- это набор синхронизированных
реплик. Что ж, это подмножество реплик, которые на данный момент живы и догнали лидер.
Запустите Producer для отправки сообщений
Эта процедура остается такой же, как и при настройке с одним брокером.
Example
bin/kafka-console-producer.sh --broker-list localhost:9092
--topic Multibrokerapplication
Output
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
Запустить потребителя для получения сообщений
Эта процедура остается такой же, как показано в настройке с одним брокером.
Example
bin/kafka-console-consumer.sh --zookeeper localhost:2181
—topic Multibrokerapplica-tion --from-beginning
Output
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 Cluster. Теперь давайте изменим созданную тему, используя следующую команду
Syntax
bin/kafka-topics.sh —zookeeper localhost:2181 --alter --topic topic_name
--parti-tions count
Example
We have already created a topic “Hello-Kafka” with single partition count and one replica factor.
Now using “alter” command we have changed the partition count.
bin/kafka-topics.sh --zookeeper localhost:2181
--alter --topic Hello-kafka --parti-tions 2
Output
WARNING: If partitions are increased for a topic that has a key,
the partition logic or ordering of the messages will be affected
Adding partitions succeeded!
Удаление темы
Чтобы удалить тему, вы можете использовать следующий синтаксис.
Syntax
bin/kafka-topics.sh --zookeeper localhost:2181 --delete --topic topic_name
Example
bin/kafka-topics.sh --zookeeper localhost:2181 --delete --topic Hello-kafka
Output
> Topic Hello-kafka marked for deletion
Note −Это не повлияет, если delete.topic.enable не установлено значение true
Давайте создадим приложение для публикации и использования сообщений с помощью Java-клиента. Клиент производителя Kafka состоит из следующих API.
KafkaProducer API
Давайте разберемся с наиболее важным набором API производителя Kafka в этом разделе. Центральная часть API KafkaProducer
- это
класс KafkaProducer
. Класс KafkaProducer предоставляет возможность подключить брокера Kafka в его конструкторе с помощью следующих методов.
Класс KafkaProducer предоставляет метод send для асинхронной отправки сообщений в тему. Подпись send () выглядит следующим образом
producer.send(new ProducerRecord<byte[],byte[]>(topic,
partition, key1, value1) , callback);
ProducerRecord - Производитель управляет буфером записей, ожидающих отправки.
Callback - Предоставляемый пользователем обратный вызов для выполнения, когда запись была подтверждена сервером (null указывает на отсутствие обратного вызова).
Класс KafkaProducer предоставляет метод очистки, гарантирующий, что все ранее отправленные сообщения были фактически завершены. Синтаксис метода flush следующий:
public void flush()
Класс KafkaProducer предоставляет метод partitionFor, который помогает получить метаданные раздела для заданной темы. Это можно использовать для пользовательского разбиения. Сигнатура этого метода следующая -
public Map metrics()
Он возвращает карту внутренних показателей, поддерживаемую производителем.
public void close () - класс KafkaProducer предоставляет блоки метода close до тех пор, пока не будут выполнены все ранее отправленные запросы.
API производителя
Центральной частью Producer API является класс Producer
. Класс Producer предоставляет возможность подключить брокер Kafka в своем конструкторе следующими методами.
Продюсерский класс
Класс производителя предоставляет метод отправки для send сообщения в одну или несколько тем, используя следующие подписи.
public void send(KeyedMessaget<k,v> message)
- sends the data to a single topic,par-titioned by key using either sync or async producer.
public void send(List<KeyedMessage<k,v>>messages)
- sends data to multiple topics.
Properties prop = new Properties();
prop.put(producer.type,”async”)
ProducerConfig config = new ProducerConfig(prop);
Есть два типа производителей - Sync и Async.
Та же конфигурация API применима и к производителю Sync
. Разница между ними в том, что производитель синхронизации отправляет сообщения напрямую, но отправляет сообщения в фоновом режиме. Если вам нужна более высокая пропускная способность, предпочтительнее использовать асинхронный производитель. В предыдущих выпусках, таких как 0.8, производитель async не имел обратного вызова для send () для регистрации обработчиков ошибок. Это доступно только в текущей версии 0.9.
public void close ()
Класс производителя обеспечивает close метод, чтобы закрыть соединения пула производителей со всеми брокерами Kafka.
Настройки конфигурации
Основные параметры конфигурации Producer API перечислены в следующей таблице для лучшего понимания -
S.No | Параметры конфигурации и описание |
---|---|
1 | client.id идентифицирует приложение производителя |
2 | producer.type либо синхронизировать, либо асинхронно |
3 | acks Конфигурация acks контролирует критерии, по которым запросы производителя считаются завершенными. |
4 | retries Если запрос производителя не удается, автоматически повторите попытку с определенным значением. |
5 | bootstrap.servers начальный список брокеров. |
6 | linger.ms если вы хотите уменьшить количество запросов, вы можете установить linger.ms на значение, большее некоторого значения. |
7 | key.serializer Ключ для интерфейса сериализатора. |
8 | value.serializer значение для интерфейса сериализатора. |
9 | batch.size Размер буфера. |
10 | buffer.memory контролирует общий объем памяти, доступной производителю для буферизации. |
ProducerRecord API
ProducerRecord - это пара ключ / значение, которая отправляется в кластер Kafka. Конструктор класса ProductRecord для создания записи с парами разделов, ключей и значений с использованием следующей подписи.
public ProducerRecord (string topic, int partition, k key, v value)
Topic - определяемое пользователем название темы, которое будет добавлено к записи.
Partition - количество разделов
Key - Ключ, который будет включен в запись.
- Value - Запись содержимого
public ProducerRecord (string topic, k key, v value)
Конструктор класса ProducerRecord используется для создания записи с парами ключ, значение и без разделения.
Topic - Создайте тему для присвоения записи.
Key - ключ для записи.
Value - записывать содержимое.
public ProducerRecord (string topic, v value)
Класс ProducerRecord создает запись без раздела и ключа.
Topic - создайте тему.
Value - записывать содержимое.
Методы класса ProducerRecord перечислены в следующей таблице -
S.No | Методы и описание класса |
---|---|
1 | public string topic() Тема будет добавлена к записи. |
2 | public K key() Ключ, который будет включен в запись. Если такого ключа нет, здесь будет повторно установлено значение null. |
3 | public V value() Запишите содержимое. |
4 | partition() Количество разделов для записи |
Приложение SimpleProducer
Перед созданием приложения сначала запустите ZooKeeper и брокер Kafka, затем создайте свою собственную тему в брокере Kafka, используя команду create topic. После этого создайте класс java с именем Sim-pleProducer.java
и введите следующий код
.
//import util.properties packages
import java.util.Properties;
//import simple producer packages
import org.apache.kafka.clients.producer.Producer;
//import KafkaProducer packages
import org.apache.kafka.clients.producer.KafkaProducer;
//import ProducerRecord packages
import org.apache.kafka.clients.producer.ProducerRecord;
//Create java class named “SimpleProducer”
public class SimpleProducer {
public static void main(String[] args) throws Exception{
// Check arguments length value
if(args.length == 0){
System.out.println("Enter topic name”);
return;
}
//Assign topicName to string variable
String topicName = args[0].toString();
// create instance for properties to access producer configs
Properties props = new Properties();
//Assign localhost id
props.put("bootstrap.servers", “localhost:9092");
//Set acknowledgements for producer requests.
props.put("acks", “all");
//If the request fails, the producer can automatically retry,
props.put("retries", 0);
//Specify buffer size in config
props.put("batch.size", 16384);
//Reduce the no of requests less than 0
props.put("linger.ms", 1);
//The buffer.memory controls the total amount of memory available to the producer for buffering.
props.put("buffer.memory", 33554432);
props.put("key.serializer",
"org.apache.kafka.common.serializa-tion.StringSerializer");
props.put("value.serializer",
"org.apache.kafka.common.serializa-tion.StringSerializer");
Producer<String, String> producer = new KafkaProducer
<String, String>(props);
for(int i = 0; i < 10; i++)
producer.send(new ProducerRecord<String, String>(topicName,
Integer.toString(i), Integer.toString(i)));
System.out.println(“Message sent successfully”);
producer.close();
}
}
Compilation - Приложение можно скомпилировать с помощью следующей команды.
javac -cp “/path/to/kafka/kafka_2.11-0.9.0.0/lib/*” *.java
Execution - Приложение можно запустить с помощью следующей команды.
java -cp “/path/to/kafka/kafka_2.11-0.9.0.0/lib/*”:. SimpleProducer <topic-name>
Output
Message sent successfully
To check the above output open new terminal and type Consumer CLI command to receive messages.
>> bin/kafka-console-consumer.sh --zookeeper localhost:2181 —topic <topic-name> —from-beginning
1
2
3
4
5
6
7
8
9
10
Простой потребительский пример
На данный момент мы создали производителя для отправки сообщений в кластер Kafka. Теперь давайте создадим потребителя для приема сообщений из кластера Kafka. KafkaConsumer API используется для получения сообщений из кластера Kafka. Конструктор класса KafkaConsumer определен ниже.
public KafkaConsumer(java.util.Map<java.lang.String,java.lang.Object> configs)
configs - Вернуть карту потребительских конфигов.
Класс KafkaConsumer имеет следующие важные методы, перечисленные в таблице ниже.
S.No | Метод и описание |
---|---|
1 | public java.util.Set<TopicPar-tition> assignment() Получите набор разделов, назначенных в данный момент потребителем. |
2 | public string subscription() Подпишитесь на указанный список тем, чтобы получать разделы с динамической подписью. |
3 | public void sub-scribe(java.util.List<java.lang.String> topics, ConsumerRe-balanceListener listener) Подпишитесь на указанный список тем, чтобы получать разделы с динамической подписью. |
4 | public void unsubscribe() Отписаться от тем из данного списка разделов. |
5 | public void sub-scribe(java.util.List<java.lang.String> topics) Подпишитесь на указанный список тем, чтобы получать разделы с динамической подписью. Если данный список тем пуст, он обрабатывается так же, как unsubscribe (). |
6 | public void sub-scribe(java.util.regex.Pattern pattern, ConsumerRebalanceLis-tener listener) Шаблон аргумента относится к шаблону подписки в формате регулярного выражения, а аргумент слушателя получает уведомления от шаблона подписки. |
7 | public void as-sign(java.util.List<TopicParti-tion> partitions) Вручную назначьте заказчику список разделов. |
8 | poll() Получение данных для указанных тем или разделов с помощью одного из API подписки / назначения. Это вернет ошибку, если темы не подписаны до опроса данных. |
9 | public void commitSync() Смещения фиксации, возвращенные в последнем опросе () для всего подписанного списка тем и разделов. Та же операция применяется к commitAsyn (). |
10 | public void seek(TopicPartition partition, long offset) Получите текущее значение смещения, которое потребитель будет использовать в следующем методе poll (). |
11 | public void resume() Возобновить приостановленные разделы. |
12 | public void wakeup() Разбуди потребителя. |
ConsumerRecord API
API ConsumerRecord используется для получения записей из кластера Kafka. Этот API состоит из имени темы, номера раздела, из которого принимается запись, и смещения, указывающего на запись в разделе Kafka. Класс ConsumerRecord используется для создания записи потребителя с определенным названием темы, количеством разделов и парами <ключ, значение>. Имеет следующую подпись.
public ConsumerRecord(string topic,int partition, long offset,K key, V value)
Topic - Название темы для записи потребителя, полученной от кластера Kafka.
Partition - Раздел по теме.
Key - Ключ записи, если ключ не существует, возвращается null.
Value - Запись содержимого.
ConsumerRecords API
ConsumerRecords API действует как контейнер для ConsumerRecord. Этот API используется для хранения списка ConsumerRecord для каждого раздела по определенной теме. Его конструктор определен ниже.
public ConsumerRecords(java.util.Map<TopicPartition,java.util.List
<Consumer-Record>K,V>>> records)
TopicPartition - Вернуть карту раздела по определенной теме.
Records - Вернуть список ConsumerRecord.
В классе ConsumerRecords определены следующие методы.
S.No | Методы и описание |
---|---|
1 | public int count() Количество записей по всем темам. |
2 | public Set partitions() Набор разделов с данными в этом наборе записей (если данные не были возвращены, набор пуст). |
3 | public Iterator iterator() Итератор позволяет циклически перебирать, получать или перемещать элементы. |
4 | public List records() Получить список записей для данного раздела. |
Настройки конфигурации
Параметры конфигурации для основных параметров конфигурации Consumer client API перечислены ниже:
S.No | Настройки и описание |
---|---|
1 | bootstrap.servers Загрузочный список брокеров. |
2 | group.id Включает отдельного потребителя в группу. |
3 | enable.auto.commit Включите автоматическую фиксацию для смещений, если значение истинно, в противном случае не зафиксировано. |
4 | auto.commit.interval.ms Возвращает, как часто обновленные потребляемые смещения записываются в ZooKeeper. |
5 | session.timeout.ms Указывает, сколько миллисекунд Kafka будет ждать, пока ZooKeeper ответит на запрос (чтение или запись), прежде чем отказаться и продолжить прием сообщений. |
Приложение SimpleConsumer
Шаги приложения производителя здесь остаются прежними. Сначала запустите своего брокера ZooKeeper и Kafka. Затем создайте приложение SimpleConsumer
с классом java с именем SimpleCon-sumer.java
и введите следующий код.
import java.util.Properties;
import java.util.Arrays;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.ConsumerRecord;
public class SimpleConsumer {
public static void main(String[] args) throws Exception {
if(args.length == 0){
System.out.println("Enter topic name");
return;
}
//Kafka consumer configuration settings
String topicName = args[0].toString();
Properties props = new Properties();
props.put("bootstrap.servers", "localhost:9092");
props.put("group.id", "test");
props.put("enable.auto.commit", "true");
props.put("auto.commit.interval.ms", "1000");
props.put("session.timeout.ms", "30000");
props.put("key.deserializer",
"org.apache.kafka.common.serializa-tion.StringDeserializer");
props.put("value.deserializer",
"org.apache.kafka.common.serializa-tion.StringDeserializer");
KafkaConsumer<String, String> consumer = new KafkaConsumer
<String, String>(props);
//Kafka Consumer subscribes list of topics here.
consumer.subscribe(Arrays.asList(topicName))
//print the topic name
System.out.println("Subscribed to topic " + topicName);
int i = 0;
while (true) {
ConsumerRecords<String, String> records = con-sumer.poll(100);
for (ConsumerRecord<String, String> record : records)
// print the offset,key and value for the consumer records.
System.out.printf("offset = %d, key = %s, value = %s\n",
record.offset(), record.key(), record.value());
}
}
}
Compilation - Приложение можно скомпилировать с помощью следующей команды.
javac -cp “/path/to/kafka/kafka_2.11-0.9.0.0/lib/*” *.java
Execution − Приложение можно запустить с помощью следующей команды
java -cp “/path/to/kafka/kafka_2.11-0.9.0.0/lib/*”:. SimpleConsumer <topic-name>
Input- Откройте интерфейс командной строки производителя и отправьте несколько сообщений в тему. Вы можете ввести небольшой ввод как «Hello Consumer».
Output - Ниже будет вывод.
Subscribed to topic Hello-Kafka
offset = 3, key = null, value = Hello Consumer
Группа потребителей - это многопоточное или многопоточное потребление из тем Kafka.
Потребительская группа
Потребители могут присоединиться к группе, используя один и тот же
group.id.
Максимальный параллелизм группы состоит в том, что количество потребителей в группе ← количество разделов.
Kafka назначает разделы темы потребителю в группе, так что каждый раздел используется ровно одним потребителем в группе.
Kafka гарантирует, что сообщение будет прочитано только одним потребителем в группе.
Потребители могут видеть сообщения в том порядке, в котором они были сохранены в журнале.
Ребалансировка потребителя
Добавление большего количества процессов / потоков приведет к перебалансировке Kafka. Если какой-либо потребитель или брокер не может отправить контрольный сигнал ZooKeeper, его можно перенастроить через кластер Kafka. Во время этой перебалансировки Kafka назначит доступные разделы доступным потокам, возможно, переместив раздел в другой процесс.
import java.util.Properties;
import java.util.Arrays;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.ConsumerRecord;
public class ConsumerGroup {
public static void main(String[] args) throws Exception {
if(args.length < 2){
System.out.println("Usage: consumer <topic> <groupname>");
return;
}
String topic = args[0].toString();
String group = args[1].toString();
Properties props = new Properties();
props.put("bootstrap.servers", "localhost:9092");
props.put("group.id", group);
props.put("enable.auto.commit", "true");
props.put("auto.commit.interval.ms", "1000");
props.put("session.timeout.ms", "30000");
props.put("key.deserializer",
"org.apache.kafka.common.serialization.ByteArraySerializer");
props.put("value.deserializer",
"org.apache.kafka.common.serializa-tion.StringDeserializer");
KafkaConsumer<String, String> consumer = new KafkaConsumer<String, String>(props);
consumer.subscribe(Arrays.asList(topic));
System.out.println("Subscribed to topic " + topic);
int i = 0;
while (true) {
ConsumerRecords<String, String> records = consumer.poll(100);
for (ConsumerRecord<String, String> record : records)
System.out.printf("offset = %d, key = %s, value = %s\n",
record.offset(), record.key(), record.value());
}
}
}
Компиляция
javac -cp “/path/to/kafka/kafka_2.11-0.9.0.0/libs/*" ConsumerGroup.java
Исполнение
>>java -cp “/path/to/kafka/kafka_2.11-0.9.0.0/libs/*":.
ConsumerGroup <topic-name> my-group
>>java -cp "/home/bala/Workspace/kafka/kafka_2.11-0.9.0.0/libs/*":.
ConsumerGroup <topic-name> my-group
Здесь мы создали образец названия группы как my-group
с двумя потребителями. Точно так же вы можете создать свою группу и количество потребителей в группе.
Ввод
Откройте интерфейс командной строки производителя и отправьте несколько сообщений вроде -
Test consumer group 01
Test consumer group 02
Выход первого процесса
Subscribed to topic Hello-kafka
offset = 3, key = null, value = Test consumer group 01
Выход второго процесса
Subscribed to topic Hello-kafka
offset = 3, key = null, value = Test consumer group 02
Надеюсь, вы бы поняли SimpleConsumer и ConsumeGroup, используя демонстрацию клиента Java. Теперь у вас есть представление о том, как отправлять и получать сообщения с помощью клиента Java. Давайте продолжим интеграцию Kafka с технологиями больших данных в следующей главе.
В этой главе мы узнаем, как интегрировать Kafka с Apache Storm.
О шторме
Первоначально Storm был создан Натаном Марцем и командой BackType. За короткое время Apache Storm стал стандартом для распределенной системы обработки в реальном времени, позволяющей обрабатывать огромные объемы данных. Storm работает очень быстро, и тест показал, что он обрабатывает более миллиона кортежей в секунду на узел. Apache Storm работает непрерывно, потребляя данные из настроенных источников (Spouts) и передает данные по конвейеру обработки (Bolts). Вместе изливы и болты образуют топологию.
Интеграция со Storm
Kafka и Storm естественным образом дополняют друг друга, и их мощное сотрудничество позволяет выполнять потоковую аналитику в реальном времени для быстро перемещающихся больших данных. Интеграция Kafka и Storm призвана упростить разработчикам прием и публикацию потоков данных из топологий Storm.
Концептуальный поток
Носик - это источник ручьев. Например, носик может считывать кортежи из темы Kafka и выдавать их в виде потока. Болт потребляет входные потоки, обрабатывает и, возможно, испускает новые потоки. Bolts могут делать что угодно: запускать функции, фильтровать кортежи, выполнять агрегирование потоковой передачи, объединения потоковой передачи, общаться с базами данных и многое другое. Каждый узел в топологии Storm выполняется параллельно. Топология работает бесконечно, пока вы ее не отключите. Storm автоматически переназначит все неудачные задачи. Кроме того, Storm гарантирует, что данные не будут потеряны, даже если машины выйдут из строя и сообщения будут сброшены.
Давайте подробно рассмотрим API интеграции Kafka-Storm. Есть три основных класса для интеграции Kafka с Storm. Они следующие -
BrokerHosts - ZkHosts и StaticHosts
BrokerHosts - это интерфейс, а ZkHosts и StaticHosts - две его основные реализации. ZkHosts используется для динамического отслеживания брокеров Kafka, сохраняя детали в ZooKeeper, в то время как StaticHosts используется для ручной / статической установки брокеров Kafka и их деталей. ZkHosts - это простой и быстрый способ получить доступ к брокеру Kafka.
Подпись ZkHosts выглядит следующим образом -
public ZkHosts(String brokerZkStr, String brokerZkPath)
public ZkHosts(String brokerZkStr)
Где brokerZkStr - это хост ZooKeeper, а brokerZkPath - это путь ZooKeeper для хранения деталей брокера Kafka.
KafkaConfig API
Этот API используется для определения параметров конфигурации для кластера Kafka. Подпись Кафки Кон-фига определяется следующим образом
public KafkaConfig(BrokerHosts hosts, string topic)
Hosts - BrokerHosts может быть ZkHosts / StaticHosts.
Topic - название темы.
SpoutConfig API
Spoutconfig - это расширение KafkaConfig, которое поддерживает дополнительную информацию ZooKeeper.
public SpoutConfig(BrokerHosts hosts, string topic, string zkRoot, string id)
Hosts - BrokerHosts может быть любой реализацией интерфейса BrokerHosts.
Topic - название темы.
zkRoot - Корневой путь ZooKeeper.
id −Носик хранит состояние смещений, использованных в Zookeeper. Идентификатор должен однозначно идентифицировать ваш носик.
SchemeAsMultiScheme
SchemeAsMultiScheme - это интерфейс, который определяет, как ByteBuffer, полученный из Kafka, преобразуется в штормовой кортеж. Он является производным от MultiScheme и принимает реализацию класса Scheme. Существует множество реализаций класса Scheme, и одна из таких реализаций - StringScheme, которая анализирует байт как простую строку. Он также управляет именованием вашего поля вывода. Подпись определяется следующим образом.
public SchemeAsMultiScheme(Scheme scheme)
Scheme - байтовый буфер из кафки.
KafkaSpout API
KafkaSpout - это наша реализация spout, которая будет интегрирована со Storm. Он извлекает сообщения из темы kafka и передает их в экосистему Storm в виде кортежей. KafkaSpout получает информацию о конфигурации из SpoutConfig.
Ниже приведен пример кода для создания простого носика Kafka.
// ZooKeeper connection string
BrokerHosts hosts = new ZkHosts(zkConnString);
//Creating SpoutConfig Object
SpoutConfig spoutConfig = new SpoutConfig(hosts,
topicName, "/" + topicName UUID.randomUUID().toString());
//convert the ByteBuffer to String.
spoutConfig.scheme = new SchemeAsMultiScheme(new StringScheme());
//Assign SpoutConfig to KafkaSpout.
KafkaSpout kafkaSpout = new KafkaSpout(spoutConfig);
Создание болта
Bolt - это компонент, который принимает кортежи в качестве входных данных, обрабатывает кортежи и создает новые кортежи в качестве выходных. Bolts будет реализовывать интерфейс IRichBolt. В этой программе для выполнения операций используются два класса болтов WordSplitter-Bolt и WordCounterBolt.
Интерфейс IRichBolt имеет следующие методы -
Prepare- Предоставляет болту среду для выполнения. Исполнители запустят этот метод для инициализации носика.
Execute - Обработка одного кортежа ввода.
Cleanup - Вызывается при отключении болта.
declareOutputFields - Объявляет схему вывода кортежа.
Давайте создадим SplitBolt.java, который реализует логику для разделения предложения на слова и CountBolt.java, который реализует логику для разделения уникальных слов и подсчета их появления.
SplitBolt.java
import java.util.Map;
import backtype.storm.tuple.Tuple;
import backtype.storm.tuple.Fields;
import backtype.storm.tuple.Values;
import backtype.storm.task.OutputCollector;
import backtype.storm.topology.OutputFieldsDeclarer;
import backtype.storm.topology.IRichBolt;
import backtype.storm.task.TopologyContext;
public class SplitBolt implements IRichBolt {
private OutputCollector collector;
@Override
public void prepare(Map stormConf, TopologyContext context,
OutputCollector collector) {
this.collector = collector;
}
@Override
public void execute(Tuple input) {
String sentence = input.getString(0);
String[] words = sentence.split(" ");
for(String word: words) {
word = word.trim();
if(!word.isEmpty()) {
word = word.toLowerCase();
collector.emit(new Values(word));
}
}
collector.ack(input);
}
@Override
public void declareOutputFields(OutputFieldsDeclarer declarer) {
declarer.declare(new Fields("word"));
}
@Override
public void cleanup() {}
@Override
public Map<String, Object> getComponentConfiguration() {
return null;
}
}
CountBolt.java
import java.util.Map;
import java.util.HashMap;
import backtype.storm.tuple.Tuple;
import backtype.storm.task.OutputCollector;
import backtype.storm.topology.OutputFieldsDeclarer;
import backtype.storm.topology.IRichBolt;
import backtype.storm.task.TopologyContext;
public class CountBolt implements IRichBolt{
Map<String, Integer> counters;
private OutputCollector collector;
@Override
public void prepare(Map stormConf, TopologyContext context,
OutputCollector collector) {
this.counters = new HashMap<String, Integer>();
this.collector = collector;
}
@Override
public void execute(Tuple input) {
String str = input.getString(0);
if(!counters.containsKey(str)){
counters.put(str, 1);
}else {
Integer c = counters.get(str) +1;
counters.put(str, c);
}
collector.ack(input);
}
@Override
public void cleanup() {
for(Map.Entry<String, Integer> entry:counters.entrySet()){
System.out.println(entry.getKey()+" : " + entry.getValue());
}
}
@Override
public void declareOutputFields(OutputFieldsDeclarer declarer) {
}
@Override
public Map<String, Object> getComponentConfiguration() {
return null;
}
}
Отправка в топологию
Топология Storm - это, по сути, структура Thrift. Класс TopologyBuilder предоставляет простые и легкие методы для создания сложных топологий. Класс TopologyBuilder имеет методы для установки носика (setSpout) и установки болта (setBolt). Наконец, TopologyBuilder имеет createTopology для создания топологии. Методы shuffleGrouping и fieldsGrouping помогают настроить группировку потоков для носика и болтов.
Local Cluster- В целях разработки мы можем создать локальный кластер с помощью
объекта LocalCluster,
а затем отправить топологию с помощью
метода submitTopology
класса LocalCluster
.
KafkaStormSample.java
import backtype.storm.Config;
import backtype.storm.LocalCluster;
import backtype.storm.topology.TopologyBuilder;
import java.util.ArrayList;
import java.util.List;
import java.util.UUID;
import backtype.storm.spout.SchemeAsMultiScheme;
import storm.kafka.trident.GlobalPartitionInformation;
import storm.kafka.ZkHosts;
import storm.kafka.Broker;
import storm.kafka.StaticHosts;
import storm.kafka.BrokerHosts;
import storm.kafka.SpoutConfig;
import storm.kafka.KafkaConfig;
import storm.kafka.KafkaSpout;
import storm.kafka.StringScheme;
public class KafkaStormSample {
public static void main(String[] args) throws Exception{
Config config = new Config();
config.setDebug(true);
config.put(Config.TOPOLOGY_MAX_SPOUT_PENDING, 1);
String zkConnString = "localhost:2181";
String topic = "my-first-topic";
BrokerHosts hosts = new ZkHosts(zkConnString);
SpoutConfig kafkaSpoutConfig = new SpoutConfig (hosts, topic, "/" + topic,
UUID.randomUUID().toString());
kafkaSpoutConfig.bufferSizeBytes = 1024 * 1024 * 4;
kafkaSpoutConfig.fetchSizeBytes = 1024 * 1024 * 4;
kafkaSpoutConfig.forceFromStart = true;
kafkaSpoutConfig.scheme = new SchemeAsMultiScheme(new StringScheme());
TopologyBuilder builder = new TopologyBuilder();
builder.setSpout("kafka-spout", new KafkaSpout(kafkaSpoutCon-fig));
builder.setBolt("word-spitter", new SplitBolt()).shuffleGroup-ing("kafka-spout");
builder.setBolt("word-counter", new CountBolt()).shuffleGroup-ing("word-spitter");
LocalCluster cluster = new LocalCluster();
cluster.submitTopology("KafkaStormSample", config, builder.create-Topology());
Thread.sleep(10000);
cluster.shutdown();
}
}
Перед переносом компиляции для интеграции Kakfa-Storm требуется кураторская клиентская java-библиотека ZooKeeper. Куратор версии 2.9.1 поддерживает Apache Storm версии 0.9.5 (которую мы используем в этом руководстве). Загрузите указанные ниже файлы jar и поместите их в путь к классу java.
- curator-client-2.9.1.jar
- curator-framework-2.9.1.jar
После включения файлов зависимостей скомпилируйте программу, используя следующую команду:
javac -cp "/path/to/Kafka/apache-storm-0.9.5/lib/*" *.java
Исполнение
Запустите интерфейс командной строки Kafka Producer (объяснено в предыдущей главе), создайте новую тему с именем my-first-topic
и предоставьте несколько примеров сообщений, как показано ниже -
hello
kafka
storm
spark
test message
another test message
Теперь запустите приложение, используя следующую команду -
java -cp “/path/to/Kafka/apache-storm-0.9.5/lib/*”:. KafkaStormSample
Пример вывода этого приложения указан ниже -
storm : 1
test : 2
spark : 1
another : 1
kafka : 1
hello : 1
message : 2
В этой главе мы обсудим, как интегрировать Apache Kafka с Spark Streaming API.
О Spark
Spark Streaming API обеспечивает масштабируемую, высокопроизводительную и отказоустойчивую потоковую обработку потоков данных в реальном времени. Данные могут быть получены из многих источников, таких как Kafka, Flume, Twitter и т. Д., И могут быть обработаны с использованием сложных алгоритмов, таких как высокоуровневые функции, такие как map, reduce, join и window. Наконец, обработанные данные могут быть отправлены в файловые системы, базы данных и живые информационные панели. Устойчивые распределенные наборы данных (RDD) - это фундаментальная структура данных Spark. Это неизменяемая распределенная коллекция объектов. Каждый набор данных в RDD разделен на логические разделы, которые могут быть вычислены на разных узлах кластера.
Интеграция со Spark
Kafka - это потенциальная платформа для обмена сообщениями и интеграции для потоковой передачи Spark. Kafka действует как центральный узел для потоков данных в реальном времени и обрабатывается с использованием сложных алгоритмов в Spark Streaming. После обработки данных Spark Streaming может публиковать результаты в еще одной теме Kafka или хранить их в HDFS, базах данных или информационных панелях. На следующей диаграмме показан концептуальный поток.
Теперь давайте подробно рассмотрим API Kafka-Spark.
SparkConf API
Он представляет собой конфигурацию для приложения Spark. Используется для установки различных параметров Spark в виде пар ключ-значение.
Класс SparkConf
имеет следующие методы -
set(string key, string value) - установить переменную конфигурации.
remove(string key) - удалить ключ из конфигурации.
setAppName(string name) - установить название приложения для вашего приложения.
get(string key) - получить ключ
StreamingContext API
Это основная точка входа для функциональности Spark. SparkContext представляет подключение к кластеру Spark и может использоваться для создания RDD, аккумуляторов и широковещательных переменных в кластере. Подпись определяется, как показано ниже.
public StreamingContext(String master, String appName, Duration batchDuration,
String sparkHome, scala.collection.Seq<String> jars,
scala.collection.Map<String,String> environment)
master - URL-адрес кластера для подключения (например, mesos: // host: port, spark: // host: port, local [4]).
appName - название вашей работы, которое будет отображаться в веб-интерфейсе кластера
batchDuration - временной интервал, через который потоковые данные будут разделены на пакеты
public StreamingContext(SparkConf conf, Duration batchDuration)
Создайте StreamingContext, предоставив конфигурацию, необходимую для нового SparkContext.
conf - Параметры искры
batchDuration - временной интервал, через который потоковые данные будут разделены на пакеты
KafkaUtils API
KafkaUtils API используется для подключения кластера Kafka к потоковой передаче Spark. В этом API есть сигнатура метода createStream,
определенная ниже.
public static ReceiverInputDStream<scala.Tuple2<String,String>> createStream(
StreamingContext ssc, String zkQuorum, String groupId,
scala.collection.immutable.Map<String,Object> topics, StorageLevel storageLevel)
Показанный выше метод используется для создания входного потока, который извлекает сообщения от Kafka Brokers.
ssc - Объект StreamingContext.
zkQuorum - Кворум Zookeeper.
groupId - Идентификатор группы для этого потребителя.
topics - вернуть карту тем для использования.
storageLevel - Уровень хранения, используемый для хранения полученных объектов.
У KafkaUtils API есть другой метод createDirectStream, который используется для создания входного потока, который напрямую извлекает сообщения от Kafka Brokers без использования какого-либо получателя. Этот поток может гарантировать, что каждое сообщение от Kafka будет включено в преобразования ровно один раз.
Пример приложения написан на Scala. Чтобы скомпилировать приложение, загрузите и установите sbt
, инструмент сборки scala (аналогично maven). Ниже представлен основной код приложения.
import java.util.HashMap
import org.apache.kafka.clients.producer.{KafkaProducer, ProducerConfig, Produc-erRecord}
import org.apache.spark.SparkConf
import org.apache.spark.streaming._
import org.apache.spark.streaming.kafka._
object KafkaWordCount {
def main(args: Array[String]) {
if (args.length < 4) {
System.err.println("Usage: KafkaWordCount <zkQuorum><group> <topics> <numThreads>")
System.exit(1)
}
val Array(zkQuorum, group, topics, numThreads) = args
val sparkConf = new SparkConf().setAppName("KafkaWordCount")
val ssc = new StreamingContext(sparkConf, Seconds(2))
ssc.checkpoint("checkpoint")
val topicMap = topics.split(",").map((_, numThreads.toInt)).toMap
val lines = KafkaUtils.createStream(ssc, zkQuorum, group, topicMap).map(_._2)
val words = lines.flatMap(_.split(" "))
val wordCounts = words.map(x => (x, 1L))
.reduceByKeyAndWindow(_ + _, _ - _, Minutes(10), Seconds(2), 2)
wordCounts.print()
ssc.start()
ssc.awaitTermination()
}
}
Сценарий сборки
Интеграция Spark-Kafka зависит от искры, потока искры и Jar интеграции Spark. Создайте новый файл build.sbt
и укажите сведения о приложении и его зависимости. При компиляции и упаковке приложения sbt
загрузит необходимый jar-файл.
name := "Spark Kafka Project"
version := "1.0"
scalaVersion := "2.10.5"
libraryDependencies += "org.apache.spark" %% "spark-core" % "1.6.0"
libraryDependencies += "org.apache.spark" %% "spark-streaming" % "1.6.0"
libraryDependencies += "org.apache.spark" %% "spark-streaming-kafka" % "1.6.0"
Составление / упаковка
Выполните следующую команду, чтобы скомпилировать и упаковать файл jar приложения. Нам нужно отправить файл jar в консоль Spark для запуска приложения.
sbt package
Отправка в Spark
Запустите интерфейс командной строки Kafka Producer (объяснено в предыдущей главе), создайте новую тему с именем my-first-topic
и предоставьте несколько примеров сообщений, как показано ниже.
Another spark test message
Выполните следующую команду, чтобы отправить приложение в Spark console.
/usr/local/spark/bin/spark-submit --packages org.apache.spark:spark-streaming
-kafka_2.10:1.6.0 --class "KafkaWordCount" --master local[4] target/scala-2.10/spark
-kafka-project_2.10-1.0.jar localhost:2181 <group name> <topic name> <number of threads>
Пример вывода этого приложения показан ниже.
spark console messages ..
(Test,1)
(spark,1)
(another,1)
(message,1)
spark console message ..
Давайте проанализируем приложение в реальном времени, чтобы получить последние ленты Twitter и его хэштеги. Ранее мы видели интеграцию Storm и Spark с Kafka. В обоих сценариях мы создали Kafka Producer (используя cli) для отправки сообщения в экосистему Kafka. Затем интеграция шторма и искры считывает сообщения с помощью потребителя Kafka и вводит их в экосистему шторма и искры соответственно. Итак, практически нам нужно создать Kafka Producer, который должен:
- Читайте ленты Twitter, используя «Twitter Streaming API»,
- Обработка фидов,
- Извлеките хэштеги и
- Отправь Кафке.
Как только Kafka
получает хэштеги
, интеграция Storm / Spark получает информацию и отправляет ее в экосистему Storm / Spark.
Twitter Streaming API
Доступ к «Twitter Streaming API» можно получить на любом языке программирования. «Twitter4j» - это неофициальная библиотека Java с открытым исходным кодом, которая предоставляет модуль на основе Java для простого доступа к «API потоковой передачи Twitter». Twitter4j предоставляет фреймворк на основе слушателя для доступа к твитам. Чтобы получить доступ к «API потоковой передачи Twitter», нам необходимо войти в учетную запись разработчика Twitter и получить следующиеOAuth детали аутентификации.
- Customerkey
- CustomerSecret
- AccessToken
- AccessTookenSecret
После создания учетной записи разработчика загрузите файлы jar «twitter4j» и поместите их в путь класса java.
Полный код производителя Twitter Kafka (KafkaTwitterProducer.java) указан ниже -
import java.util.Arrays;
import java.util.Properties;
import java.util.concurrent.LinkedBlockingQueue;
import twitter4j.*;
import twitter4j.conf.*;
import org.apache.kafka.clients.producer.Producer;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerRecord;
public class KafkaTwitterProducer {
public static void main(String[] args) throws Exception {
LinkedBlockingQueue<Status> queue = new LinkedBlockingQueue<Sta-tus>(1000);
if(args.length < 5){
System.out.println(
"Usage: KafkaTwitterProducer <twitter-consumer-key>
<twitter-consumer-secret> <twitter-access-token>
<twitter-access-token-secret>
<topic-name> <twitter-search-keywords>");
return;
}
String consumerKey = args[0].toString();
String consumerSecret = args[1].toString();
String accessToken = args[2].toString();
String accessTokenSecret = args[3].toString();
String topicName = args[4].toString();
String[] arguments = args.clone();
String[] keyWords = Arrays.copyOfRange(arguments, 5, arguments.length);
ConfigurationBuilder cb = new ConfigurationBuilder();
cb.setDebugEnabled(true)
.setOAuthConsumerKey(consumerKey)
.setOAuthConsumerSecret(consumerSecret)
.setOAuthAccessToken(accessToken)
.setOAuthAccessTokenSecret(accessTokenSecret);
TwitterStream twitterStream = new TwitterStreamFactory(cb.build()).get-Instance();
StatusListener listener = new StatusListener() {
@Override
public void onStatus(Status status) {
queue.offer(status);
// System.out.println("@" + status.getUser().getScreenName()
+ " - " + status.getText());
// System.out.println("@" + status.getUser().getScreen-Name());
/*for(URLEntity urle : status.getURLEntities()) {
System.out.println(urle.getDisplayURL());
}*/
/*for(HashtagEntity hashtage : status.getHashtagEntities()) {
System.out.println(hashtage.getText());
}*/
}
@Override
public void onDeletionNotice(StatusDeletionNotice statusDeletion-Notice) {
// System.out.println("Got a status deletion notice id:"
+ statusDeletionNotice.getStatusId());
}
@Override
public void onTrackLimitationNotice(int numberOfLimitedStatuses) {
// System.out.println("Got track limitation notice:" +
num-berOfLimitedStatuses);
}
@Override
public void onScrubGeo(long userId, long upToStatusId) {
// System.out.println("Got scrub_geo event userId:" + userId +
"upToStatusId:" + upToStatusId);
}
@Override
public void onStallWarning(StallWarning warning) {
// System.out.println("Got stall warning:" + warning);
}
@Override
public void onException(Exception ex) {
ex.printStackTrace();
}
};
twitterStream.addListener(listener);
FilterQuery query = new FilterQuery().track(keyWords);
twitterStream.filter(query);
Thread.sleep(5000);
//Add Kafka producer config settings
Properties props = new Properties();
props.put("bootstrap.servers", "localhost:9092");
props.put("acks", "all");
props.put("retries", 0);
props.put("batch.size", 16384);
props.put("linger.ms", 1);
props.put("buffer.memory", 33554432);
props.put("key.serializer",
"org.apache.kafka.common.serializa-tion.StringSerializer");
props.put("value.serializer",
"org.apache.kafka.common.serializa-tion.StringSerializer");
Producer<String, String> producer = new KafkaProducer<String, String>(props);
int i = 0;
int j = 0;
while(i < 10) {
Status ret = queue.poll();
if (ret == null) {
Thread.sleep(100);
i++;
}else {
for(HashtagEntity hashtage : ret.getHashtagEntities()) {
System.out.println("Hashtag: " + hashtage.getText());
producer.send(new ProducerRecord<String, String>(
top-icName, Integer.toString(j++), hashtage.getText()));
}
}
}
producer.close();
Thread.sleep(5000);
twitterStream.shutdown();
}
}
Компиляция
Скомпилируйте приложение, используя следующую команду -
javac -cp “/path/to/kafka/libs/*”:”/path/to/twitter4j/lib/*”:. KafkaTwitterProducer.java
Исполнение
Откройте две консоли. Запустите скомпилированное приложение, как показано ниже, в одной консоли.
java -cp “/path/to/kafka/libs/*”:”/path/to/twitter4j/lib/*”:
. KafkaTwitterProducer <twitter-consumer-key>
<twitter-consumer-secret>
<twitter-access-token>
<twitter-ac-cess-token-secret>
my-first-topic food
Запустите любое из приложений Spark / Storm, описанных в предыдущей главе, в другом окне. Главное отметить, что используемая тема должна быть одинаковой в обоих случаях. Здесь мы использовали «my-first-topic» в качестве названия темы.
Вывод
Вывод этого приложения будет зависеть от ключевых слов и текущей ленты твиттера. Пример вывода указан ниже (интеграция шторма).
. . .
food : 1
foodie : 2
burger : 1
. . .
Инструмент Kafka из пакета org.apache.kafka.tools. *. Инструменты подразделяются на системные инструменты и инструменты репликации.
Системные инструменты
Системные инструменты можно запустить из командной строки с помощью сценария запуска класса. Синтаксис следующий -
bin/kafka-run-class.sh package.class - - options
Некоторые из системных инструментов упомянуты ниже -
Kafka Migration Tool - Этот инструмент используется для переноса брокера с одной версии на другую.
Mirror Maker - Этот инструмент используется для обеспечения зеркального отображения одного кластера Kafka в другой.
Consumer Offset Checker - Этот инструмент отображает группу потребителей, тему, разделы, смещение, размер журнала, владельца для указанного набора тем и группу потребителей.
Инструмент репликации
Репликация Kafka - это инструмент проектирования высокого уровня. Целью добавления инструмента репликации является повышение надежности и доступности. Некоторые из инструментов репликации упомянуты ниже -
Create Topic Tool - Это создает тему с количеством разделов по умолчанию, коэффициентом репликации и использует схему Kafka по умолчанию для назначения реплик.
List Topic Tool- Этот инструмент отображает информацию по заданному списку тем. Если в командной строке не указаны темы, инструмент запрашивает Zookeeper, чтобы получить все темы и перечисляет информацию для них. Поля, отображаемые инструментом, - это имя темы, раздел, лидер, реплики, isr.
Add Partition Tool- Создание темы, необходимо указать количество разделов для темы. В дальнейшем для темы могут потребоваться дополнительные разделы, когда объем темы увеличится. Этот инструмент помогает добавить больше разделов для определенной темы, а также позволяет вручную назначать реплики добавленных разделов.
Kafka поддерживает многие из лучших на сегодняшний день промышленных приложений. В этой главе мы дадим очень краткий обзор некоторых наиболее заметных приложений Kafka.
Twitter - это онлайн-служба социальной сети, которая предоставляет платформу для отправки и получения пользовательских твитов. Зарегистрированные пользователи могут читать и публиковать твиты, а незарегистрированные пользователи могут только читать твиты. Twitter использует Storm-Kafka как часть своей инфраструктуры обработки потоковых данных.
Apache Kafka используется в LinkedIn для данных потока активности и операционных показателей. Система обмена сообщениями Kafka помогает LinkedIn с различными продуктами, такими как LinkedIn Newsfeed, LinkedIn Today, для использования онлайн-сообщений, а также с системами офлайн-аналитики, такими как Hadoop. Высокая надежность Kafka также является одним из ключевых факторов в связи с LinkedIn.
Netflix
Netflix - американский многонациональный поставщик потокового мультимедиа в Интернете по запросу. Netflix использует Kafka для мониторинга и обработки событий в реальном времени.
Mozilla
Mozilla - это сообщество свободного программного обеспечения, созданное в 1998 году членами Netscape. Kafka скоро заменит часть текущей производственной системы Mozilla для сбора данных о производительности и использовании из браузера конечного пользователя для таких проектов, как Telemetry, Test Pilot и т. Д.
Oracle
Oracle обеспечивает встроенное подключение к Kafka из своего продукта Enterprise Service Bus под названием OSB (Oracle Service Bus), который позволяет разработчикам использовать встроенные посреднические возможности OSB для реализации поэтапных конвейеров данных.