실시간 신청 (Twitter)

실시간 애플리케이션을 분석하여 최신 트위터 피드와 해시 태그를 가져 오겠습니다. 이전에 Storm 및 Spark와 Kafka의 통합을 확인했습니다. 두 시나리오에서 Kafka 생태계에 메시지를 보내기 위해 Kafka Producer (cli 사용)를 만들었습니다. 그런 다음 스톰 및 스파크 통합은 Kafka 소비자를 사용하여 메시지를 읽고이를 각각 스톰 및 스파크 생태계에 주입합니다. 그래서 실제로 우리는 Kafka Producer를 만들어야합니다.

"Twitter Streaming API"를 사용하여 트위터 피드를 읽고,
피드 처리,
HashTag를 추출하고
Kafka에게 보내십시오.

한 번 해시 태그는 카프카, 폭풍에 의해 수신 / 스파크의 통합은 INFOR-mation의를 받고 스톰 / 스파크 생태계에 보낼 수 있습니다.

Twitter 스트리밍 API

"Twitter Streaming API"는 모든 프로그래밍 언어로 액세스 할 수 있습니다. "twitter4j"는 "Twitter Streaming API"에 쉽게 액세스 할 수있는 Java 기반 모듈을 제공하는 오픈 소스 비공식 Java 라이브러리입니다. "twitter4j"는 트윗에 액세스 할 수있는 리스너 기반 프레임 워크를 제공합니다. "Twitter Streaming API"에 액세스하려면 Twitter 개발자 계정에 로그인해야하며 다음을 받아야합니다.OAuth 인증 세부 사항.

Customerkey
CustomerSecret
AccessToken
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("@" + 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();
   }
}

편집

다음 명령을 사용하여 응용 프로그램을 컴파일하십시오-

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

실행

두 개의 콘솔을 엽니 다. 하나의 콘솔에서 아래와 같이 위에서 컴파일 된 응용 프로그램을 실행합니다.

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 애플리케이션 중 하나를 실행합니다. 주목해야 할 요점은 사용 된 주제가 두 경우 모두 동일해야한다는 것입니다. 여기서는 주제 이름으로 "my-first-topic"을 사용했습니다.

산출

이 애플리케이션의 출력은 키워드와 트위터의 현재 피드에 따라 달라집니다. 아래에 샘플 출력이 지정되어 있습니다 (폭풍 통합).

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