Gerçek Zamanlı Uygulama (Twitter)

En son twitter beslemelerini ve hashtag'lerini almak için gerçek zamanlı bir uygulamayı analiz edelim. Daha önce Storm ve Spark'ın Kafka ile entegrasyonunu görmüştük. Her iki senaryoda da, Kafka ekosistemine mesaj göndermek için bir Kafka Yapımcısı (cli kullanarak) oluşturduk. Ardından, fırtına ve kıvılcım entegrasyonu, Kafka tüketicisini kullanarak mesajları okur ve sırasıyla fırtına ve kıvılcım ekosistemine enjekte eder. Yani, pratik olarak bir Kafka Yapımcısı yaratmalıyız ki bu -

  • Twitter akışlarını "Twitter Akış API" kullanarak okuyun,
  • Beslemeleri işleyin,
  • HashTag'leri ayıklayın ve
  • Kafka'ya gönder.

Bir kez Hashtag'ler Kafka, Fırtına tarafından alınır / Kıvılcım entegrasyon bulu-mation almak ve Fırtına / Kıvılcım ekosistem gönderin.

Twitter Akış API'si

"Twitter Streaming API" herhangi bir programlama dilinde erişilebilir. “Twitter4j”, “Twitter Streaming API” ye kolayca erişmek için Java tabanlı bir modül sağlayan açık kaynaklı, resmi olmayan bir Java kitaplığıdır. "Twitter4j", tweet'lere erişmek için dinleyici tabanlı bir çerçeve sağlar. "Twitter Streaming API" ye erişmek için, Twitter geliştirici hesabına giriş yapmalı ve aşağıdakileri almalıyızOAuth kimlik doğrulama ayrıntıları.

  • Customerkey
  • CustomerSecret
  • AccessToken
  • AccessTookenSecret

Geliştirici hesabı oluşturulduktan sonra, "twitter4j" jar dosyalarını indirin ve java sınıf yoluna yerleştirin.

Tam Twitter Kafka yapımcı kodlaması (KafkaTwitterProducer.java) aşağıda listelenmiştir -

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

Derleme

Aşağıdaki komutu kullanarak uygulamayı derleyin -

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

Yürütme

İki konsol açın. Yukarıda derlenen uygulamayı aşağıda gösterildiği gibi bir konsolda çalıştırın.

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

Önceki bölümde başka bir win-dow'da açıklanan Spark / Storm uygulamasından herhangi birini çalıştırın. Dikkat edilmesi gereken en önemli nokta, kullanılan konunun her iki durumda da aynı olması gerektiğidir. Burada konu adı olarak “ilk-konu” kullandık.

Çıktı

Bu uygulamanın çıktısı, anahtar kelimelere ve twitter'ın mevcut beslemesine bağlı olacaktır. Aşağıda örnek bir çıktı belirtilmiştir (fırtına entegrasyonu).

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