Apache Kafka - Интеграция со Storm

В этой главе мы узнаем, как интегрировать Kafka с Apache Storm.

О шторме

Первоначально Storm был создан Натаном Марцем и командой BackType. За короткое время Apache Storm стал стандартом для распределенной системы обработки в реальном времени, позволяющей обрабатывать огромные объемы данных. Storm работает очень быстро, и тест показал, что он обрабатывается более миллиона кортежей в секунду на узел. Apache Storm работает непрерывно, потребляя данные из настроенных источников (Spout) и передает данные по конвейеру обработки (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. Подпись Kafka Con-fig определяется следующим образом

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