Apache Kafka - Integration mit Sturm

In diesem Kapitel erfahren Sie, wie Sie Kafka in Apache Storm integrieren.

Über Sturm

Storm wurde ursprünglich von Nathan Marz und dem Team von BackType erstellt. In kurzer Zeit wurde Apache Storm zum Standard für verteilte Echtzeitverarbeitungssysteme, mit denen Sie ein großes Datenvolumen verarbeiten können. Storm ist sehr schnell und wurde von einem Benchmark mit über einer Million Tupeln pro Sekunde und Knoten verarbeitet. Apache Storm wird kontinuierlich ausgeführt, verbraucht Daten aus den konfigurierten Quellen (Spouts) und leitet die Daten über die Verarbeitungspipeline (Bolts) weiter. Kombiniert bilden Ausgüsse und Schrauben eine Topologie.

Integration mit Storm

Kafka und Storm ergänzen sich auf natürliche Weise, und ihre leistungsstarke Zusammenarbeit ermöglicht Echtzeit-Streaming-Analysen für sich schnell bewegende Big Data. Die Integration von Kafka und Storm soll Entwicklern das Aufnehmen und Veröffentlichen von Datenströmen aus Storm-Topologien erleichtern.

Konzeptioneller Ablauf

Ein Auslauf ist eine Quelle von Strömen. Zum Beispiel kann ein Auslauf Tupel von einem Kafka-Thema lesen und sie als Stream ausgeben. Ein Bolzen verbraucht Eingabestreams, verarbeitet sie und gibt möglicherweise neue Streams aus. Bolts können alles tun, von der Ausführung von Funktionen über das Filtern von Tupeln bis hin zu Streaming-Aggregationen, Streaming-Joins, Gesprächen mit Datenbanken und vielem mehr. Jeder Knoten in einer Storm-Topologie wird parallel ausgeführt. Eine Topologie wird unbegrenzt ausgeführt, bis Sie sie beenden. Storm weist fehlgeschlagene Aufgaben automatisch neu zu. Darüber hinaus garantiert Storm, dass kein Datenverlust auftritt, selbst wenn die Computer ausfallen und Nachrichten gelöscht werden.

Lassen Sie uns die Kafka-Storm-Integrations-APIs im Detail durchgehen. Es gibt drei Hauptklassen, um Kafka in Storm zu integrieren. Sie sind wie folgt -

BrokerHosts - ZkHosts & StaticHosts

BrokerHosts ist eine Schnittstelle und ZkHosts und StaticHosts sind die beiden Hauptimplementierungen. ZkHosts wird verwendet, um die Kafka-Broker dynamisch zu verfolgen, indem die Details in ZooKeeper verwaltet werden, während StaticHosts verwendet wird, um die Kafka-Broker und ihre Details manuell / statisch festzulegen. ZkHosts ist der einfache und schnelle Weg, um auf den Kafka-Broker zuzugreifen.

Die Signatur von ZkHosts lautet wie folgt:

public ZkHosts(String brokerZkStr, String brokerZkPath)
public ZkHosts(String brokerZkStr)

Dabei ist BrokerZkStr der ZooKeeper-Host und BrokerZkPath der ZooKeeper-Pfad zum Verwalten der Kafka-Brokerdetails.

KafkaConfig API

Diese API wird verwendet, um Konfigurationseinstellungen für den Kafka-Cluster zu definieren. Die Signatur von Kafka Con-fig ist wie folgt definiert

public KafkaConfig(BrokerHosts hosts, string topic)

Hosts - Die BrokerHosts können ZkHosts / StaticHosts sein.

Topic - Themenname.

SpoutConfig API

Spoutconfig ist eine Erweiterung von KafkaConfig, die zusätzliche ZooKeeper-Informationen unterstützt.

public SpoutConfig(BrokerHosts hosts, string topic, string zkRoot, string id)

Hosts - Die BrokerHosts können eine beliebige Implementierung der BrokerHosts-Schnittstelle sein
Topic - Themenname.
zkRoot - ZooKeeper-Stammpfad.
id −Der Auslauf speichert den Status der Offsets, die in Zookeeper verbraucht werden. Die ID sollte Ihren Auslauf eindeutig identifizieren.

SchemeAsMultiScheme

SchemeAsMultiScheme ist eine Schnittstelle, die vorschreibt, wie der von Kafka verbrauchte ByteBuffer in ein Sturmtupel umgewandelt wird. Es ist von MultiScheme abgeleitet und akzeptiert die Implementierung der Scheme-Klasse. Es gibt viele Implementierungen der Scheme-Klasse und eine solche Implementierung ist StringScheme, die das Byte als einfachen String analysiert. Es steuert auch die Benennung Ihres Ausgabefeldes. Die Signatur ist wie folgt definiert.

public SchemeAsMultiScheme(Scheme scheme)

Scheme - Byte-Puffer aus Kafka verbraucht.

KafkaSpout API

KafkaSpout ist unsere Auslaufimplementierung, die in Storm integriert wird. Es holt die Nachrichten aus dem Kafka-Thema und gibt sie als Tupel an das Storm-Ökosystem ab. KafkaSpout erhält seine Konfigurationsdetails von SpoutConfig.

Unten finden Sie einen Beispielcode zum Erstellen eines einfachen Kafka-Auslaufs.

// 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);

Bolzenerstellung

Bolt ist eine Komponente, die Tupel als Eingabe verwendet, das Tupel verarbeitet und neue Tupel als Ausgabe erzeugt. Bolzen implementieren die IRichBolt-Schnittstelle. In diesem Programm werden zwei Schraubenklassen WordSplitter-Bolt und WordCounterBolt verwendet, um die Operationen auszuführen.

Die IRichBolt-Schnittstelle verfügt über die folgenden Methoden:

Prepare- Versorgt den Bolzen mit einer auszuführenden Umgebung. Die Ausführenden führen diese Methode aus, um den Auslauf zu initialisieren.
Execute - Verarbeiten Sie ein einzelnes Tupel der Eingabe.
Cleanup - Wird aufgerufen, wenn ein Bolzen abgeschaltet wird.
declareOutputFields - Deklariert das Ausgabeschema des Tupels.

Erstellen wir SplitBolt.java, das die Logik zum Aufteilen eines Satzes in Wörter implementiert, und CountBolt.java, das die Logik implementiert, um eindeutige Wörter zu trennen und deren Auftreten zu zählen.

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;
   }
}

Senden an die Topologie

Die Storm-Topologie ist im Grunde eine Thrift-Struktur. Die TopologyBuilder-Klasse bietet einfache Methoden zum Erstellen komplexer Topologien. Die TopologyBuilder-Klasse verfügt über Methoden zum Setzen von Ausgüssen (setSpout) und zum Setzen von Bolzen (setBolt). Schließlich verfügt TopologyBuilder über createTopology zum Erstellen von to-pology. shuffleGrouping- und FelderGrouping-Methoden helfen beim Festlegen der Stream-Gruppierung für Auslauf und Schrauben.

Local Cluster- Zu Entwicklungszwecken können wir mit dem LocalCluster- Objekt einen lokalen Cluster erstellen und dann die Topologie mit der submitTopology- Methode der LocalCluster- Klasse senden .

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();
   }
}

Vor dem Verschieben der Kompilierung benötigt die Kakfa-Storm-Integration die Java-Bibliothek des Kurators ZooKeeper. Curator Version 2.9.1 unterstützt Apache Storm Version 0.9.5 (die wir in diesem Tutorial verwenden). Laden Sie die unten angegebenen JAR-Dateien herunter und platzieren Sie sie im Java-Klassenpfad.

curator-client-2.9.1.jar
curator-framework-2.9.1.jar

Kompilieren Sie das Programm nach dem Einfügen von Abhängigkeitsdateien mit dem folgenden Befehl:

javac -cp "/path/to/Kafka/apache-storm-0.9.5/lib/*" *.java

Ausführung

Starten Sie die Kafka Producer CLI (im vorherigen Kapitel erläutert), erstellen Sie ein neues Thema mit dem Namen " Mein erstes Thema" und geben Sie einige Beispielnachrichten an, wie unten gezeigt.

hello
kafka
storm
spark
test message
another test message

Führen Sie nun die Anwendung mit dem folgenden Befehl aus:

java -cp “/path/to/Kafka/apache-storm-0.9.5/lib/*”:. KafkaStormSample

Die Beispielausgabe dieser Anwendung ist unten angegeben -

storm : 1
test : 2
spark : 1
another : 1
kafka : 1
hello : 1
message : 2