Apache Storm - Arbeitsbeispiel

Wir haben die technischen Details des Apache Storm durchgearbeitet und jetzt ist es an der Zeit, einige einfache Szenarien zu codieren.

Szenario - Mobile Call Log Analyzer

Der Mobilanruf und seine Dauer werden als Eingabe für Apache Storm angegeben, und der Storm verarbeitet und gruppiert den Anruf zwischen demselben Anrufer und Empfänger und deren Gesamtzahl.

Auslauferstellung

Auslauf ist eine Komponente, die zur Datengenerierung verwendet wird. Grundsätzlich implementiert ein Auslauf eine IRichSpout-Schnittstelle. Die "IRichSpout" -Schnittstelle verfügt über die folgenden wichtigen Methoden:

open- Versorgt den Auslauf mit einer auszuführenden Umgebung. Die Ausführenden führen diese Methode aus, um den Auslauf zu initialisieren.
nextTuple - Gibt die generierten Daten über den Kollektor aus.
close - Diese Methode wird aufgerufen, wenn ein Auslauf heruntergefahren wird.
declareOutputFields - Deklariert das Ausgabeschema des Tupels.
ack - Bestätigt, dass ein bestimmtes Tupel verarbeitet wird
fail - Gibt an, dass ein bestimmtes Tupel nicht verarbeitet und nicht erneut verarbeitet werden soll.

Öffnen

Die Unterschrift des open Methode ist wie folgt -

open(Map conf, TopologyContext context, SpoutOutputCollector collector)

conf - Bietet eine Sturmkonfiguration für diesen Auslauf.
context - Bietet vollständige Informationen über die Auslaufstelle in der Topologie, ihre Aufgaben-ID sowie Eingabe- und Ausgabeinformationen.
collector - Ermöglicht es uns, das Tupel auszugeben, das von den Schrauben verarbeitet wird.

nextTuple

Die Unterschrift des nextTuple Methode ist wie folgt -

nextTuple()

nextTuple () wird periodisch aus derselben Schleife wie die Methoden ack () und fail () aufgerufen. Es muss die Kontrolle über den Thread freigeben, wenn keine Arbeit zu erledigen ist, damit die anderen Methoden aufgerufen werden können. In der ersten Zeile von nextTuple wird also überprüft, ob die Verarbeitung abgeschlossen ist. In diesem Fall sollte es mindestens eine Millisekunde lang schlafen, um die Belastung des Prozessors zu verringern, bevor Sie zurückkehren.

schließen

Die Unterschrift des close Methode ist wie folgt -

close()

declareOutputFields

Die Unterschrift des declareOutputFields Methode ist wie folgt -

declareOutputFields(OutputFieldsDeclarer declarer)

declarer - Es wird verwendet, um Ausgabestream-IDs, Ausgabefelder usw. zu deklarieren.

Diese Methode wird verwendet, um das Ausgabeschema des Tupels anzugeben.

ack

Die Unterschrift des ack Methode ist wie folgt -

ack(Object msgId)

Diese Methode bestätigt, dass ein bestimmtes Tupel verarbeitet wurde.

Scheitern

Die Unterschrift des nextTuple Methode ist wie folgt -

ack(Object msgId)

Diese Methode informiert, dass ein bestimmtes Tupel nicht vollständig verarbeitet wurde. Storm wird das spezifische Tupel erneut verarbeiten.

FakeCallLogReaderSpout

In unserem Szenario müssen wir die Anrufprotokolldetails erfassen. Die Informationen des Anrufprotokolls enthalten.

Anrufernummer
Empfängernummer
duration

Da wir keine Echtzeitinformationen zu Anrufprotokollen haben, werden gefälschte Anrufprotokolle erstellt. Die gefälschten Informationen werden mit der Zufallsklasse erstellt. Der vollständige Programmcode ist unten angegeben.

Codierung - FakeCallLogReaderSpout.java

import java.util.*;
//import storm tuple packages
import backtype.storm.tuple.Fields;
import backtype.storm.tuple.Values;

//import Spout interface packages
import backtype.storm.topology.IRichSpout;
import backtype.storm.topology.OutputFieldsDeclarer;
import backtype.storm.spout.SpoutOutputCollector;
import backtype.storm.task.TopologyContext;

//Create a class FakeLogReaderSpout which implement IRichSpout interface 
   to access functionalities
	
public class FakeCallLogReaderSpout implements IRichSpout {
   //Create instance for SpoutOutputCollector which passes tuples to bolt.
   private SpoutOutputCollector collector;
   private boolean completed = false;
	
   //Create instance for TopologyContext which contains topology data.
   private TopologyContext context;
	
   //Create instance for Random class.
   private Random randomGenerator = new Random();
   private Integer idx = 0;

   @Override
   public void open(Map conf, TopologyContext context, SpoutOutputCollector collector) {
      this.context = context;
      this.collector = collector;
   }

   @Override
   public void nextTuple() {
      if(this.idx <= 1000) {
         List<String> mobileNumbers = new ArrayList<String>();
         mobileNumbers.add("1234123401");
         mobileNumbers.add("1234123402");
         mobileNumbers.add("1234123403");
         mobileNumbers.add("1234123404");

         Integer localIdx = 0;
         while(localIdx++ < 100 && this.idx++ < 1000) {
            String fromMobileNumber = mobileNumbers.get(randomGenerator.nextInt(4));
            String toMobileNumber = mobileNumbers.get(randomGenerator.nextInt(4));
				
            while(fromMobileNumber == toMobileNumber) {
               toMobileNumber = mobileNumbers.get(randomGenerator.nextInt(4));
            }
				
            Integer duration = randomGenerator.nextInt(60);
            this.collector.emit(new Values(fromMobileNumber, toMobileNumber, duration));
         }
      }
   }

   @Override
   public void declareOutputFields(OutputFieldsDeclarer declarer) {
      declarer.declare(new Fields("from", "to", "duration"));
   }

   //Override all the interface methods
   @Override
   public void close() {}

   public boolean isDistributed() {
      return false;
   }

   @Override
   public void activate() {}

   @Override 
   public void deactivate() {}

   @Override
   public void ack(Object msgId) {}

   @Override
   public void fail(Object msgId) {}

   @Override
   public Map<String, Object> getComponentConfiguration() {
      return null;
   }
}

Bolzenerstellung

Bolt ist eine Komponente, die Tupel als Eingabe verwendet, das Tupel verarbeitet und neue Tupel als Ausgabe erzeugt. Schrauben werden implementiertIRichBoltSchnittstelle. In diesem Programm zwei SchraubenklassenCallLogCreatorBolt und CallLogCounterBolt werden 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.

Bereiten

Die Unterschrift des prepare Methode ist wie folgt -

prepare(Map conf, TopologyContext context, OutputCollector collector)

conf - Bietet Storm-Konfiguration für diesen Bolzen.
context - Bietet vollständige Informationen über die Position der Schraube in der Topologie, ihre Aufgaben-ID, Eingabe- und Ausgabeinformationen usw.
collector - Ermöglicht es uns, das verarbeitete Tupel auszugeben.

ausführen

Die Unterschrift des execute Methode ist wie folgt -

execute(Tuple tuple)

Hier tuple ist das zu verarbeitende Eingabetupel.

Das executeMethode verarbeitet jeweils ein einzelnes Tupel. Auf die Tupeldaten kann mit der Methode getValue der Tupelklasse zugegriffen werden. Es ist nicht erforderlich, das Eingabetupel sofort zu verarbeiten. Mehrere Tupel können als ein einziges Ausgabetupel verarbeitet und ausgegeben werden. Das verarbeitete Tupel kann mithilfe der OutputCollector-Klasse ausgegeben werden.

Aufräumen

Die Unterschrift des cleanup Methode ist wie folgt -

cleanup()

declareOutputFields

Die Unterschrift des declareOutputFields Methode ist wie folgt -

declareOutputFields(OutputFieldsDeclarer declarer)

Hier der Parameter declarer wird verwendet, um Ausgabestream-IDs, Ausgabefelder usw. zu deklarieren.

Diese Methode wird verwendet, um das Ausgabeschema des Tupels anzugeben

Call Log Creator Bolt

Die Anrufprotokoll-Erstellungsschraube empfängt das Anrufprotokoll-Tupel. Das Anrufprotokolltupel enthält Anrufernummer, Empfängernummer und Anrufdauer. Diese Schraube erzeugt einfach einen neuen Wert, indem sie die Anrufernummer und die Empfängernummer kombiniert. Das Format des neuen Werts lautet "Anrufernummer - Empfängernummer" und wird als neues Feld "Anruf" bezeichnet. Der vollständige Code ist unten angegeben.

Codierung - CallLogCreatorBolt.java

//import util packages
import java.util.HashMap;
import java.util.Map;

import backtype.storm.tuple.Fields;
import backtype.storm.tuple.Values;
import backtype.storm.task.OutputCollector;
import backtype.storm.task.TopologyContext;

//import Storm IRichBolt package
import backtype.storm.topology.IRichBolt;
import backtype.storm.topology.OutputFieldsDeclarer;
import backtype.storm.tuple.Tuple;

//Create a class CallLogCreatorBolt which implement IRichBolt interface
public class CallLogCreatorBolt implements IRichBolt {
   //Create instance for OutputCollector which collects and emits tuples to produce output
   private OutputCollector collector;

   @Override
   public void prepare(Map conf, TopologyContext context, OutputCollector collector) {
      this.collector = collector;
   }

   @Override
   public void execute(Tuple tuple) {
      String from = tuple.getString(0);
      String to = tuple.getString(1);
      Integer duration = tuple.getInteger(2);
      collector.emit(new Values(from + " - " + to, duration));
   }

   @Override
   public void cleanup() {}

   @Override
   public void declareOutputFields(OutputFieldsDeclarer declarer) {
      declarer.declare(new Fields("call", "duration"));
   }
	
   @Override
   public Map<String, Object> getComponentConfiguration() {
      return null;
   }
}

Rufprotokoll Zählerschraube

Die Anrufprotokollzählerschraube empfängt den Anruf und seine Dauer als Tupel. Diese Schraube initialisiert ein Wörterbuchobjekt (Map-Objekt) in der Vorbereitungsmethode. ImexecuteMethode überprüft es das Tupel und erstellt einen neuen Eintrag im Wörterbuchobjekt für jeden neuen "Aufruf" -Wert im Tupel und setzt einen Wert 1 im Wörterbuchobjekt. Für den bereits verfügbaren Eintrag im Wörterbuch wird nur der Wert erhöht. In einfachen Worten, diese Schraube speichert den Aufruf und seine Anzahl im Wörterbuchobjekt. Anstatt den Aufruf und seine Anzahl im Wörterbuch zu speichern, können wir ihn auch in einer Datenquelle speichern. Der vollständige Programmcode lautet wie folgt:

Codierung - CallLogCounterBolt.java

import java.util.HashMap;
import java.util.Map;

import backtype.storm.tuple.Fields;
import backtype.storm.tuple.Values;
import backtype.storm.task.OutputCollector;
import backtype.storm.task.TopologyContext;
import backtype.storm.topology.IRichBolt;
import backtype.storm.topology.OutputFieldsDeclarer;
import backtype.storm.tuple.Tuple;

public class CallLogCounterBolt implements IRichBolt {
   Map<String, Integer> counterMap;
   private OutputCollector collector;

   @Override
   public void prepare(Map conf, TopologyContext context, OutputCollector collector) {
      this.counterMap = new HashMap<String, Integer>();
      this.collector = collector;
   }

   @Override
   public void execute(Tuple tuple) {
      String call = tuple.getString(0);
      Integer duration = tuple.getInteger(1);
		
      if(!counterMap.containsKey(call)){
         counterMap.put(call, 1);
      }else{
         Integer c = counterMap.get(call) + 1;
         counterMap.put(call, c);
      }
		
      collector.ack(tuple);
   }

   @Override
   public void cleanup() {
      for(Map.Entry<String, Integer> entry:counterMap.entrySet()){
         System.out.println(entry.getKey()+" : " + entry.getValue());
      }
   }

   @Override
   public void declareOutputFields(OutputFieldsDeclarer declarer) {
      declarer.declare(new Fields("call"));
   }
	
   @Override
   public Map<String, Object> getComponentConfiguration() {
      return null;
   }
	
}

Topologie erstellen

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 Festlegen des Auslaufs(setSpout) und Bolzen setzen (setBolt). Schließlich verfügt TopologyBuilder über createTopology, um eine Topologie zu erstellen. Verwenden Sie das folgende Codefragment, um eine Topologie zu erstellen:

TopologyBuilder builder = new TopologyBuilder();

builder.setSpout("call-log-reader-spout", new FakeCallLogReaderSpout());

builder.setBolt("call-log-creator-bolt", new CallLogCreatorBolt())
   .shuffleGrouping("call-log-reader-spout");

builder.setBolt("call-log-counter-bolt", new CallLogCounterBolt())
   .fieldsGrouping("call-log-creator-bolt", new Fields("call"));

shuffleGrouping und fieldsGrouping Methoden helfen beim Festlegen der Stream-Gruppierung für Auslauf und Schrauben.

Lokaler Cluster

Zu Entwicklungszwecken können wir einen lokalen Cluster mit dem Objekt "LocalCluster" erstellen und dann die Topologie mit der Methode "submitTopology" der Klasse "LocalCluster" senden. Eines der Argumente für "submitTopology" ist eine Instanz der Klasse "Config". Die Klasse "Config" wird verwendet, um Konfigurationsoptionen festzulegen, bevor die Topologie gesendet wird. Diese Konfigurationsoption wird zur Laufzeit mit der Clusterkonfiguration zusammengeführt und mit der Vorbereitungsmethode an alle Aufgaben (Auslauf und Schraube) gesendet. Sobald die Topologie an den Cluster gesendet wurde, warten wir 10 Sekunden, bis der Cluster die übermittelte Topologie berechnet hat, und fahren den Cluster dann mithilfe der Methode "shutdown" von "LocalCluster" herunter. Der vollständige Programmcode lautet wie folgt:

Codierung - LogAnalyserStorm.java

import backtype.storm.tuple.Fields;
import backtype.storm.tuple.Values;

//import storm configuration packages
import backtype.storm.Config;
import backtype.storm.LocalCluster;
import backtype.storm.topology.TopologyBuilder;

//Create main class LogAnalyserStorm submit topology.
public class LogAnalyserStorm {
   public static void main(String[] args) throws Exception{
      //Create Config instance for cluster configuration
      Config config = new Config();
      config.setDebug(true);
		
      //
      TopologyBuilder builder = new TopologyBuilder();
      builder.setSpout("call-log-reader-spout", new FakeCallLogReaderSpout());

      builder.setBolt("call-log-creator-bolt", new CallLogCreatorBolt())
         .shuffleGrouping("call-log-reader-spout");

      builder.setBolt("call-log-counter-bolt", new CallLogCounterBolt())
         .fieldsGrouping("call-log-creator-bolt", new Fields("call"));
			
      LocalCluster cluster = new LocalCluster();
      cluster.submitTopology("LogAnalyserStorm", config, builder.createTopology());
      Thread.sleep(10000);
		
      //Stop the topology
		
      cluster.shutdown();
   }
}

Erstellen und Ausführen der Anwendung

Die vollständige Anwendung verfügt über vier Java-Codes. Sie sind -

FakeCallLogReaderSpout.java
CallLogCreaterBolt.java
CallLogCounterBolt.java
LogAnalyerStorm.java

Die Anwendung kann mit dem folgenden Befehl erstellt werden:

javac -cp “/path/to/storm/apache-storm-0.9.5/lib/*” *.java

Die Anwendung kann mit dem folgenden Befehl ausgeführt werden:

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

Ausgabe

Sobald die Anwendung gestartet ist, werden die vollständigen Details zum Cluster-Startprozess, zur Auslauf- und Bolzenverarbeitung und schließlich zum Cluster-Herunterfahren ausgegeben. In "CallLogCounterBolt" haben wir den Anruf und seine Zähldetails gedruckt. Diese Informationen werden auf der Konsole wie folgt angezeigt:

1234123402 - 1234123401 : 78
1234123402 - 1234123404 : 88
1234123402 - 1234123403 : 105
1234123401 - 1234123404 : 74
1234123401 - 1234123403 : 81
1234123401 - 1234123402 : 81
1234123403 - 1234123404 : 86
1234123404 - 1234123401 : 63
1234123404 - 1234123402 : 82
1234123403 - 1234123402 : 83
1234123404 - 1234123403 : 86
1234123403 - 1234123401 : 93

Nicht-JVM-Sprachen

Storm-Topologien werden über Thrift-Schnittstellen implementiert, wodurch es einfach ist, Topologien in jeder Sprache einzureichen. Storm unterstützt Ruby, Python und viele andere Sprachen. Werfen wir einen Blick auf die Python-Bindung.

Python-Bindung

Python ist eine universell interpretierte, interaktive, objektorientierte und übergeordnete Programmiersprache. Storm unterstützt Python bei der Implementierung seiner Topologie. Python unterstützt das Senden, Verankern, Bestätigen und Protokollieren von Vorgängen.

Wie Sie wissen, können Schrauben in jeder Sprache definiert werden. In einer anderen Sprache geschriebene Bolzen werden als Unterprozesse ausgeführt, und Storm kommuniziert mit diesen Unterprozessen mit JSON-Nachrichten über stdin / stdout. Nehmen Sie zuerst eine Beispielschraube WordCount, die die Python-Bindung unterstützt.

public static class WordCount implements IRichBolt {
   public WordSplit() {
      super("python", "splitword.py");
   }
	
   public void declareOutputFields(OutputFieldsDeclarer declarer) {
      declarer.declare(new Fields("word"));
   }
}

Hier die Klasse WordCount implementiert die IRichBoltSchnittstelle und Ausführung mit Python-Implementierung angegebenes Super-Methodenargument "splitword.py". Erstellen Sie nun eine Python-Implementierung mit dem Namen "splitword.py".

import storm
   class WordCountBolt(storm.BasicBolt):
      def process(self, tup):
         words = tup.values[0].split(" ")
         for word in words:
         storm.emit([word])
WordCountBolt().run()

Dies ist die Beispielimplementierung für Python, die die Wörter in einem bestimmten Satz zählt. Ebenso können Sie auch mit anderen unterstützenden Sprachen binden.