Application en temps réel (Twitter)

Analysons une application en temps réel pour obtenir les derniers flux Twitter et ses hashtags. Plus tôt, nous avons vu l'intégration de Storm et Spark avec Kafka. Dans les deux scénarios, nous avons créé un producteur Kafka (en utilisant cli) pour envoyer un message à l'écosystème Kafka. Ensuite, l'intégration tempête et étincelle lit les messages en utilisant le consommateur Kafka et l'injecte respectivement dans l'écosystème tempête et étincelle. Donc, pratiquement, nous devons créer un producteur Kafka, qui devrait -

Lisez les flux Twitter à l'aide de «Twitter Streaming API»,
Traiter les flux,
Extraire les HashTags et
Envoyez-le à Kafka.

Une fois les HashTags reçus par Kafka, l'intégration Storm / Spark reçoit les informations et les envoie à l'écosystème Storm / Spark.

API de streaming Twitter

L '«API Twitter Streaming» est accessible dans n'importe quel langage de programmation. «Twitter4j» est une bibliothèque Java non officielle open source, qui fournit un module basé sur Java pour accéder facilement à «Twitter Streaming API». Le «twitter4j» fournit un cadre basé sur l'auditeur pour accéder aux tweets. Pour accéder à «Twitter Streaming API», nous devons nous connecter au compte de développeur Twitter et obtenir les éléments suivantsOAuth détails d'authentification.

Customerkey
CustomerSecret
AccessToken
AccessTookenSecret

Une fois le compte développeur créé, téléchargez les fichiers jar «twitter4j» et placez-les dans le chemin de classe java.

Le codage complet du producteur Twitter Kafka (KafkaTwitterProducer.java) est répertorié ci-dessous -

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

Compilation

Compilez l'application à l'aide de la commande suivante -

javac -cp “/path/to/kafka/libs/*”:”/path/to/twitter4j/lib/*”:. KafkaTwitterProducer.java

Exécution

Ouvrez deux consoles. Exécutez l'application compilée ci-dessus comme indiqué ci-dessous dans une console.

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

Exécutez l'une des applications Spark / Storm expliquées dans le chapitre précédent dans une autre fenêtre. Le point principal à noter est que le sujet utilisé doit être le même dans les deux cas. Ici, nous avons utilisé «mon premier sujet» comme nom du sujet.

Production

La sortie de cette application dépendra des mots-clés et du flux actuel du twitter. Un exemple de sortie est spécifié ci-dessous (intégration de tempête).

. . .
food : 1
foodie : 2
burger : 1
. . .