Apache-kafka-real-time-application

リアルタイムアプリケーション（Twitter）

リアルタイムアプリケーションを分析して、最新のtwitterフィードとそのハッシュタグを取得しましょう。 以前、StormおよびSparkとKafkaの統合を見てきました。 どちらのシナリオでも、Kafkaエコシステムにメッセージを送信するために（cliを使用して）Kafkaプロデューサーを作成しました。 次に、ストームとスパークの統合は、Kafkaコンシューマーを使用してメッセージを読み取り、ストームとスパークのエコシステムにそれぞれ注入します。 したがって、実際には、カフカプロデューサーを作成する必要があります。

• 「Twitter Streaming API」を使用してtwitterフィードを読んでください。
• フィードを処理し、
• HashTagsを抽出し、
• カフカに送ってください。

Kafkaが「ハッシュタグ」を受信すると、Storm/Spark統合が情報を受信し、Storm/Sparkエコシステムに送信します。

Twitter Streaming API

「Twitter Streaming API」には、任意のプログラミング言語でアクセスできます。 「twitter4j」はオープンソースの非公式Javaライブラリであり、「Twitter Streaming API」に簡単にアクセスするためのJavaベースのモジュールを提供します。 「twitter4j」は、ツイートにアクセスするためのリスナーベースのフレームワークを提供します。 「Twitter Streaming API」にアクセスするには、Twitter開発者アカウントにサインインする必要があり、次の OAuth 認証の詳細を取得する必要があります。

• 顧客キー
• CustomerSecret
• アクセストークン
• AccessTookenSecret

開発者アカウントが作成されたら、「twitter4j」jarファイルをダウンロードして、javaクラスパスに配置します。

完全なTwitter Kafkaプロデューサーコーディング（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();
   }
}

編集

次のコマンドを使用してアプリケーションをコンパイルします-

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

実行

2つのコンソールを開きます。 上記のコンパイル済みアプリケーションを、以下に示すように1つのコンソールで実行します。

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

前の章で説明したSpark/Stormアプリケーションのいずれかを別のwin-dowで実行します。 注意すべき主な点は、どちらの場合も同じトピックを使用する必要があるということです。 ここでは、トピック名として「my-first-topic」を使用しています。

出力

このアプリケーションの出力は、キーワードとTwitterの現在のフィードに依存します。 サンプル出力を以下に指定します（ストーム統合）。

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