Apache Kafka 實(shí)時(shí)應(yīng)用程序(Twitter)

讓我們分析一個(gè)實(shí)時(shí)應(yīng)用程序，以獲取最新的Twitter Feed和其標(biāo)簽。 早些時(shí)候，我們已經(jīng)看到了Storm和Spark與Kafka的集成。 在這兩種情況下，我們創(chuàng)建了一個(gè)Kafka生產(chǎn)者(使用cli)向Kafka生態(tài)系統(tǒng)發(fā)送消息。 然后，storm和spark集成通過(guò)使用Kafka消費(fèi)者讀取消息，并將其分別注入到storm和spark生態(tài)系統(tǒng)中。 因此，實(shí)際上我們需要?jiǎng)?chuàng)建一個(gè)Kafka Producer，

• 使用“Twitter Streaming API"閱讀Twitter Feed，
• 處理Feeds，
• 提取HashTags
• 發(fā)送到Kafka。

一旦Kafka接收到 HashTags ，Storm / Spark集成接收到該信息并將其發(fā)送到Storm / Spark生態(tài)系統(tǒng)。

Twitter Streaming API

“Twitter Streaming API"可以使用任何編程語(yǔ)言訪(fǎng)問(wèn)。 “twitter4j"是一個(gè)開(kāi)源的非官方Java庫(kù)，它提供了一個(gè)基于Java的模塊，可以輕松訪(fǎng)問(wèn)“Twitter Streaming API"。 “twitter4j"提供了一個(gè)基于監(jiān)聽(tīng)器的框架來(lái)訪(fǎng)問(wèn)tweet。 要訪(fǎng)問(wèn)“Twitter Streaming API"，我們需要登錄Twitter開(kāi)發(fā)者帳戶(hù)，并應(yīng)獲取以下 OAuth 身份驗(yàn)證詳細(xì)信息。

• Customerkey
• CustomerSecret
• AccessToken
• AccessTookenSecret

創(chuàng)建開(kāi)發(fā)人員帳戶(hù)后，下載“twitter4j"jar文件并將其放置在java類(lèi)路徑中。

完整的Twitter Kafka生產(chǎn)者編碼(KafkaTwitterProducer.java)如下所列 -

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("@" &plus; status.getUser().getScreenName() 
               &plus; " - " &plus; status.getText());
            // System.out.println("@" &plus; 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:" 
               &plus; statusDeletionNotice.getStatusId());
         }
         
         @Override
         public void onTrackLimitationNotice(int numberOfLimitedStatuses) {
            // System.out.println("Got track limitation notice:" &plus; 
               num-berOfLimitedStatuses);
         }

         @Override
         public void onScrubGeo(long userId, long upToStatusId) {
            // System.out.println("Got scrub_geo event userId:" &plus; userId &plus; 
            "upToStatusId:" &plus; upToStatusId);
         }      
         
         @Override
         public void onStallWarning(StallWarning warning) {
            // System.out.println("Got stall warning:" &plus; 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: " &plus; hashtage.getText());
               producer.send(new ProducerRecord<String, String>(
                  top-icName, Integer.toString(j++), hashtage.getText()));
            }
         }
      }
      producer.close();
      Thread.sleep(5000);
      twitterStream.shutdown();
   }
}

匯編

使用以下命令編譯應(yīng)用程序 -

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

執(zhí)行

打開(kāi)兩個(gè)控制臺(tái)。 在一個(gè)控制臺(tái)中運(yùn)行上面編譯的應(yīng)用程序，如下所示。

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

在另一個(gè)窗口中運(yùn)行前一章中解釋的Spark / Storm應(yīng)用程序中的任何一個(gè)。 主要要注意的是，在這兩種情況下使用的主題應(yīng)該是相同的。 在這里，我們使用“我的第一主題"作為主題名稱(chēng)。

輸出

此應(yīng)用程序的輸出將取決于關(guān)鍵字和Twitter的當(dāng)前Feed。 下面指定樣本輸出(集成storm)。

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