실시간 신청 (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
. . .