Apache Storm - Exemple de travail

Nous avons parcouru les détails techniques de base d'Apache Storm et il est maintenant temps de coder des scénarios simples.

Scénario - Analyseur de journaux d'appels mobiles

L'appel mobile et sa durée seront donnés en entrée à Apache Storm et le Storm traitera et regroupera l'appel entre le même appelant et le même destinataire et leur nombre total d'appels.

Création de bec

Spout est un composant utilisé pour la génération de données. Fondamentalement, un bec implémentera une interface IRichSpout. L'interface «IRichSpout» a les méthodes importantes suivantes -

  • open- Fournit au bec un environnement à exécuter. Les exécuteurs exécuteront cette méthode pour initialiser le bec.

  • nextTuple - Émet les données générées via le collecteur.

  • close - Cette méthode est appelée lorsqu'un bec va s'arrêter.

  • declareOutputFields - Déclare le schéma de sortie du tuple.

  • ack - Reconnaît qu'un tuple spécifique est traité

  • fail - Spécifie qu'un tuple spécifique n'est pas traité et ne doit pas être retraité.

Ouvert

La signature du open la méthode est la suivante -

open(Map conf, TopologyContext context, SpoutOutputCollector collector)

  • conf - Fournit une configuration de tempête pour ce bec.

  • context - Fournit des informations complètes sur l'emplacement du bec dans la topologie, son identifiant de tâche, les informations d'entrée et de sortie.

  • collector - Permet d'émettre le tuple qui sera traité par les boulons.

nextTuple

La signature du nextTuple la méthode est la suivante -

nextTuple()

nextTuple () est appelé périodiquement à partir de la même boucle que les méthodes ack () et fail (). Il doit libérer le contrôle du thread lorsqu'il n'y a aucun travail à faire, afin que les autres méthodes aient une chance d'être appelées. Ainsi, la première ligne de nextTuple vérifie si le traitement est terminé. Si tel est le cas, il doit dormir pendant au moins une milliseconde pour réduire la charge sur le processeur avant de revenir.

proche

La signature du close la méthode est la suivante -

close()

declareOutputFields

La signature du declareOutputFields la méthode est la suivante -

declareOutputFields(OutputFieldsDeclarer declarer)

declarer - Il est utilisé pour déclarer les identifiants de flux de sortie, les champs de sortie, etc.

Cette méthode est utilisée pour spécifier le schéma de sortie du tuple.

accuser

La signature du ack la méthode est la suivante -

ack(Object msgId)

Cette méthode reconnaît qu'un tuple spécifique a été traité.

échouer

La signature du nextTuple la méthode est la suivante -

ack(Object msgId)

Cette méthode informe qu'un tuple spécifique n'a pas été entièrement traité. Storm retraitera le tuple spécifique.

FakeCallLogReaderSpout

Dans notre scénario, nous devons collecter les détails du journal des appels. Les informations du journal des appels contiennent.

  • numéro de l'appelant
  • numéro de récepteur
  • duration

Étant donné que nous n'avons pas d'informations en temps réel sur les journaux d'appels, nous allons générer de faux journaux d'appels. Les fausses informations seront créées à l'aide de la classe Random. Le code de programme complet est donné ci-dessous.

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

Création de boulons

Bolt est un composant qui prend des tuples en entrée, traite le tuple et produit de nouveaux tuples en sortie. Bolts mettra en œuvreIRichBoltinterface. Dans ce programme, deux classes de boulonsCallLogCreatorBolt et CallLogCounterBolt sont utilisés pour effectuer les opérations.

L'interface IRichBolt a les méthodes suivantes -

  • prepare- Fournit au boulon un environnement à exécuter. Les exécuteurs exécuteront cette méthode pour initialiser le bec.

  • execute - Traitez un seul tuple d'entrée.

  • cleanup - Appelé lorsqu'un verrou va s'arrêter.

  • declareOutputFields - Déclare le schéma de sortie du tuple.

Préparer

La signature du prepare la méthode est la suivante -

prepare(Map conf, TopologyContext context, OutputCollector collector)

  • conf - Fournit la configuration Storm pour ce boulon.

  • context - Fournit des informations complètes sur l'emplacement du boulon dans la topologie, son identifiant de tâche, les informations d'entrée et de sortie, etc.

  • collector - Nous permet d'émettre le tuple traité.

exécuter

La signature du execute la méthode est la suivante -

execute(Tuple tuple)

Ici tuple est le tuple d'entrée à traiter.

le executeLa méthode traite un seul tuple à la fois. Les données du tuple sont accessibles par la méthode getValue de la classe Tuple. Il n'est pas nécessaire de traiter immédiatement le tuple d'entrée. Plusieurs tuple peuvent être traités et sortis comme un seul tuple de sortie. Le tuple traité peut être émis à l'aide de la classe OutputCollector.

nettoyer

La signature du cleanup la méthode est la suivante -

cleanup()

declareOutputFields

La signature du declareOutputFields la méthode est la suivante -

declareOutputFields(OutputFieldsDeclarer declarer)

Ici le paramètre declarer est utilisé pour déclarer les identifiants de flux de sortie, les champs de sortie, etc.

Cette méthode est utilisée pour spécifier le schéma de sortie du tuple

Journal des appels Creator Bolt

Le boulon du créateur du journal des appels reçoit le tuple du journal des appels. Le tuple du journal des appels comprend le numéro de l'appelant, le numéro du destinataire et la durée de l'appel. Ce verrou crée simplement une nouvelle valeur en combinant le numéro de l'appelant et le numéro du destinataire. Le format de la nouvelle valeur est "Numéro de l'appelant - Numéro du destinataire" et il est nommé comme nouveau champ, "appel". Le code complet est donné ci-dessous.

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

Boulon du compteur du journal des appels

Le boulon du compteur du journal des appels reçoit l'appel et sa durée sous forme de tuple. Ce boulon initialise un objet dictionnaire (Map) dans la méthode prepare. Dansexecute, il vérifie le tuple et crée une nouvelle entrée dans l'objet dictionnaire pour chaque nouvelle valeur «call» dans le tuple et définit une valeur 1 dans l'objet dictionnaire. Pour l'entrée déjà disponible dans le dictionnaire, il incrémente simplement sa valeur. En termes simples, ce boulon enregistre l'appel et son décompte dans l'objet dictionnaire. Au lieu d'enregistrer l'appel et son décompte dans le dictionnaire, nous pouvons également l'enregistrer dans une source de données. Le code de programme complet est le suivant -

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

Créer une topologie

La topologie Storm est essentiellement une structure Thrift. La classe TopologyBuilder fournit des méthodes simples et faciles pour créer des topologies complexes. La classe TopologyBuilder a des méthodes pour définir spout(setSpout) et pour régler le boulon (setBolt). Enfin, TopologyBuilder a createTopology pour créer une topologie. Utilisez l'extrait de code suivant pour créer une topologie -

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 et fieldsGrouping Les méthodes aident à définir le regroupement de flux pour le bec et les boulons.

Cluster local

À des fins de développement, nous pouvons créer un cluster local à l'aide de l'objet "LocalCluster", puis soumettre la topologie en utilisant la méthode "submitTopology" de la classe "LocalCluster". Un des arguments pour "submitTopology" est une instance de la classe "Config". La classe "Config" est utilisée pour définir les options de configuration avant de soumettre la topologie. Cette option de configuration sera fusionnée avec la configuration du cluster au moment de l'exécution et envoyée à toutes les tâches (spout et bolt) avec la méthode prepare. Une fois la topologie soumise au cluster, nous attendrons 10 secondes que le cluster calcule la topologie soumise, puis nous arrêterons le cluster à l'aide de la méthode «shutdown» de «LocalCluster». Le code de programme complet est le suivant -

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

Création et exécution de l'application

L'application complète a quatre codes Java. Ils sont -

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

L'application peut être créée à l'aide de la commande suivante -

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

L'application peut être exécutée à l'aide de la commande suivante -

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

Production

Une fois l'application démarrée, elle affichera les détails complets sur le processus de démarrage du cluster, le traitement du spout et du boulon, et enfin, le processus d'arrêt du cluster. Dans "CallLogCounterBolt", nous avons imprimé l'appel et ses détails de comptage. Ces informations seront affichées sur la console comme suit -

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

Langages non JVM

Les topologies Storm sont implémentées par des interfaces Thrift, ce qui facilite la soumission de topologies dans n'importe quelle langue. Storm prend en charge Ruby, Python et de nombreux autres langages. Jetons un coup d'œil à la liaison python.

Liaison Python

Python est un langage de programmation généraliste interprété, interactif, orienté objet et de haut niveau. Storm prend en charge Python pour implémenter sa topologie. Python prend en charge les opérations d'émission, d'ancrage, d'acquittement et de journalisation.

Comme vous le savez, les boulons peuvent être définis dans n'importe quelle langue. Les boulons écrits dans un autre langage sont exécutés en tant que sous-processus et Storm communique avec ces sous-processus avec des messages JSON via stdin / stdout. Commencez par prendre un exemple de boulon WordCount qui prend en charge la liaison python.

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

Ici la classe WordCount met en œuvre le IRichBoltinterface et s'exécutant avec l'argument de super méthode spécifié par l'implémentation python "splitword.py". Créez maintenant une implémentation python nommée "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()

C'est l'exemple d'implémentation pour Python qui compte les mots dans une phrase donnée. De même, vous pouvez également vous lier à d'autres langues de prise en charge.