Apache Storm trên Twitter

Ở đây trong chương này, chúng ta sẽ thảo luận về ứng dụng thời gian thực của Apache Storm. Chúng ta sẽ xem Storm được sử dụng như thế nào trên Twitter.

Twitter

Twitter là một dịch vụ mạng xã hội trực tuyến cung cấp nền tảng để gửi và nhận các tweet của người dùng. Người dùng đã đăng ký có thể đọc và đăng tweet, nhưng người dùng chưa đăng ký chỉ có thể đọc tweet. Hashtag được sử dụng để phân loại các tweet theo từ khóa bằng cách thêm # trước từ khóa có liên quan. Bây giờ chúng ta hãy thực hiện một kịch bản thời gian thực để tìm thẻ bắt đầu bằng # được sử dụng nhiều nhất cho mỗi chủ đề.

Tạo vòi

Mục đích của spout là để mọi người gửi tweet càng sớm càng tốt. Twitter cung cấp “Twitter Streaming API”, một công cụ dựa trên dịch vụ web để truy xuất các tweet do mọi người gửi trong thời gian thực. Twitter Streaming API có thể được truy cập bằng bất kỳ ngôn ngữ lập trình nào.

twitter4j là một thư viện Java không chính thức, mã nguồn mở, cung cấp một mô-đun dựa trên Java để dễ dàng truy cập vào Twitter Streaming API. twitter4jcung cấp một khuôn khổ dựa trên người nghe để truy cập các tweet. Để truy cập API phát trực tuyến của Twitter, chúng tôi cần đăng nhập vào tài khoản nhà phát triển Twitter và sẽ nhận được các chi tiết xác thực OAuth sau.

• Customerkey
• CustomerSecret
• AccessToken
• AccessTookenSecret

Storm cung cấp một kênh twitter, TwitterSampleSpout,trong bộ khởi động của nó. Chúng tôi sẽ sử dụng nó để lấy các tweet. Vòi cần chi tiết xác thực OAuth và ít nhất một từ khóa. Vòi sẽ phát ra các tweet theo thời gian thực dựa trên các từ khóa. Mã chương trình hoàn chỉnh được cung cấp bên dưới.

Mã hóa: TwitterSampleSpout.java

import java.util.Map;
import java.util.concurrent.LinkedBlockingQueue;

import twitter4j.FilterQuery;
import twitter4j.StallWarning;
import twitter4j.Status;
import twitter4j.StatusDeletionNotice;
import twitter4j.StatusListener;

import twitter4j.TwitterStream;
import twitter4j.TwitterStreamFactory;
import twitter4j.auth.AccessToken;
import twitter4j.conf.ConfigurationBuilder;

import backtype.storm.Config;
import backtype.storm.spout.SpoutOutputCollector;

import backtype.storm.task.TopologyContext;
import backtype.storm.topology.OutputFieldsDeclarer;
import backtype.storm.topology.base.BaseRichSpout;
import backtype.storm.tuple.Fields;
import backtype.storm.tuple.Values;

import backtype.storm.utils.Utils;

@SuppressWarnings("serial")
public class TwitterSampleSpout extends BaseRichSpout {
   SpoutOutputCollector _collector;
   LinkedBlockingQueue<Status> queue = null;
   TwitterStream _twitterStream;
		
   String consumerKey;
   String consumerSecret;
   String accessToken;
   String accessTokenSecret;
   String[] keyWords;
		
   public TwitterSampleSpout(String consumerKey, String consumerSecret,
      String accessToken, String accessTokenSecret, String[] keyWords) {
         this.consumerKey = consumerKey;
         this.consumerSecret = consumerSecret;
         this.accessToken = accessToken;
         this.accessTokenSecret = accessTokenSecret;
         this.keyWords = keyWords;
   }
		
   public TwitterSampleSpout() {
      // TODO Auto-generated constructor stub
   }
		
   @Override
   public void open(Map conf, TopologyContext context,
      SpoutOutputCollector collector) {
         queue = new LinkedBlockingQueue<Status>(1000);
         _collector = collector;
         StatusListener listener = new StatusListener() {
            @Override
            public void onStatus(Status status) {
               queue.offer(status);
            }
					
            @Override
            public void onDeletionNotice(StatusDeletionNotice sdn) {}
					
            @Override
            public void onTrackLimitationNotice(int i) {}
					
            @Override
            public void onScrubGeo(long l, long l1) {}
					
            @Override
            public void onException(Exception ex) {}
					
            @Override
            public void onStallWarning(StallWarning arg0) {
               // TODO Auto-generated method stub
            }
         };
				
         ConfigurationBuilder cb = new ConfigurationBuilder();
				
         cb.setDebugEnabled(true)
            .setOAuthConsumerKey(consumerKey)
            .setOAuthConsumerSecret(consumerSecret)
            .setOAuthAccessToken(accessToken)
            .setOAuthAccessTokenSecret(accessTokenSecret);
					
         _twitterStream = new TwitterStreamFactory(cb.build()).getInstance();
         _twitterStream.addListener(listener);
				
         if (keyWords.length == 0) {
            _twitterStream.sample();
         }else {
            FilterQuery query = new FilterQuery().track(keyWords);
            _twitterStream.filter(query);
         }
   }
			
   @Override
   public void nextTuple() {
      Status ret = queue.poll();
				
      if (ret == null) {
         Utils.sleep(50);
      } else {
         _collector.emit(new Values(ret));
      }
   }
			
   @Override
   public void close() {
      _twitterStream.shutdown();
   }
			
   @Override
   public Map<String, Object> getComponentConfiguration() {
      Config ret = new Config();
      ret.setMaxTaskParallelism(1);
      return ret;
   }
			
   @Override
   public void ack(Object id) {}
			
   @Override
   public void fail(Object id) {}
			
   @Override
   public void declareOutputFields(OutputFieldsDeclarer declarer) {
      declarer.declare(new Fields("tweet"));
   }
}

Trình đọc Hashtag Bolt

Tweet phát ra bởi vòi sẽ được chuyển tiếp đến HashtagReaderBolt, sẽ xử lý tweet và phát ra tất cả các thẻ bắt đầu bằng # có sẵn. HashtagReaderBolt sử dụnggetHashTagEntitiesphương pháp được cung cấp bởi twitter4j. getHashTagEntities đọc tweet và trả về danh sách các hashtag. Mã chương trình hoàn chỉnh như sau:

Mã hóa: HashtagReaderBolt.java

import java.util.HashMap;
import java.util.Map;

import twitter4j.*;
import twitter4j.conf.*;

import backtype.storm.tuple.Fields;
import backtype.storm.tuple.Values;

import backtype.storm.task.OutputCollector;
import backtype.storm.task.TopologyContext;
import backtype.storm.topology.IRichBolt;
import backtype.storm.topology.OutputFieldsDeclarer;
import backtype.storm.tuple.Tuple;

public class HashtagReaderBolt implements IRichBolt {
   private OutputCollector collector;

   @Override
   public void prepare(Map conf, TopologyContext context, OutputCollector collector) {
      this.collector = collector;
   }

   @Override
   public void execute(Tuple tuple) {
      Status tweet = (Status) tuple.getValueByField("tweet");
      for(HashtagEntity hashtage : tweet.getHashtagEntities()) {
         System.out.println("Hashtag: " + hashtage.getText());
         this.collector.emit(new Values(hashtage.getText()));
      }
   }

   @Override
   public void cleanup() {}

   @Override
   public void declareOutputFields(OutputFieldsDeclarer declarer) {
      declarer.declare(new Fields("hashtag"));
   }
	
   @Override
   public Map<String, Object> getComponentConfiguration() {
      return null;
   }
	
}

Hashtag Counter Bolt

Thẻ bắt đầu bằng # được phát ra sẽ được chuyển tiếp đến HashtagCounterBolt. Chốt này sẽ xử lý tất cả các thẻ bắt đầu bằng # và lưu từng thẻ bắt đầu bằng # và số lượng của nó trong bộ nhớ bằng cách sử dụng đối tượng Java Map. Mã chương trình hoàn chỉnh được cung cấp bên dưới.

Mã hóa: HashtagCounterBolt.java

import java.util.HashMap;
import java.util.Map;

import backtype.storm.tuple.Fields;
import backtype.storm.tuple.Values;

import backtype.storm.task.OutputCollector;
import backtype.storm.task.TopologyContext;

import backtype.storm.topology.IRichBolt;
import backtype.storm.topology.OutputFieldsDeclarer;
import backtype.storm.tuple.Tuple;

public class HashtagCounterBolt implements IRichBolt {
   Map<String, Integer> counterMap;
   private OutputCollector collector;

   @Override
   public void prepare(Map conf, TopologyContext context, OutputCollector collector) {
      this.counterMap = new HashMap<String, Integer>();
      this.collector = collector;
   }

   @Override
   public void execute(Tuple tuple) {
      String key = tuple.getString(0);

      if(!counterMap.containsKey(key)){
         counterMap.put(key, 1);
      }else{
         Integer c = counterMap.get(key) + 1;
         counterMap.put(key, c);
      }
		
      collector.ack(tuple);
   }

   @Override
   public void cleanup() {
      for(Map.Entry<String, Integer> entry:counterMap.entrySet()){
         System.out.println("Result: " + entry.getKey()+" : " + entry.getValue());
      }
   }

   @Override
   public void declareOutputFields(OutputFieldsDeclarer declarer) {
      declarer.declare(new Fields("hashtag"));
   }
	
   @Override
   public Map<String, Object> getComponentConfiguration() {
      return null;
   }
	
}

Gửi một cấu trúc liên kết

Gửi một cấu trúc liên kết là ứng dụng chính. Cấu trúc liên kết Twitter bao gồmTwitterSampleSpout, HashtagReaderBoltvà HashtagCounterBolt. Mã chương trình sau đây cho thấy cách gửi một cấu trúc liên kết.

Mã hóa: TwitterHashtagStorm.java

import java.util.*;

import backtype.storm.tuple.Fields;
import backtype.storm.tuple.Values;
import backtype.storm.Config;
import backtype.storm.LocalCluster;
import backtype.storm.topology.TopologyBuilder;

public class TwitterHashtagStorm {
   public static void main(String[] args) throws Exception{
      String consumerKey = args[0];
      String consumerSecret = args[1];
		
      String accessToken = args[2];
      String accessTokenSecret = args[3];
		
      String[] arguments = args.clone();
      String[] keyWords = Arrays.copyOfRange(arguments, 4, arguments.length);
		
      Config config = new Config();
      config.setDebug(true);
		
      TopologyBuilder builder = new TopologyBuilder();
      builder.setSpout("twitter-spout", new TwitterSampleSpout(consumerKey,
         consumerSecret, accessToken, accessTokenSecret, keyWords));

      builder.setBolt("twitter-hashtag-reader-bolt", new HashtagReaderBolt())
         .shuffleGrouping("twitter-spout");

      builder.setBolt("twitter-hashtag-counter-bolt", new HashtagCounterBolt())
         .fieldsGrouping("twitter-hashtag-reader-bolt", new Fields("hashtag"));
			
      LocalCluster cluster = new LocalCluster();
      cluster.submitTopology("TwitterHashtagStorm", config,
         builder.createTopology());
      Thread.sleep(10000);
      cluster.shutdown();
   }
}

Xây dựng và chạy ứng dụng

Ứng dụng hoàn chỉnh có bốn mã Java. Chúng như sau:

• TwitterSampleSpout.java
• HashtagReaderBolt.java
• HashtagCounterBolt.java
• TwitterHashtagStorm.java

Bạn có thể biên dịch ứng dụng bằng lệnh sau:

javac -cp “/path/to/storm/apache-storm-0.9.5/lib/*”:”/path/to/twitter4j/lib/*” *.java

Thực thi ứng dụng bằng các lệnh sau:

javac -cp “/path/to/storm/apache-storm-0.9.5/lib/*”:”/path/to/twitter4j/lib/*”:.
TwitterHashtagStorm <customerkey> <customersecret> <accesstoken> <accesstokensecret>
<keyword1> <keyword2> … <keywordN>

Đầu ra

Ứng dụng sẽ in hashtag hiện có sẵn và số lượng của nó. Đầu ra phải tương tự như sau:

Result: jazztastic : 1
Result: foodie : 1
Result: Redskins : 1
Result: Recipe : 1
Result: cook : 1
Result: android : 1
Result: food : 2
Result: NoToxicHorseMeat : 1
Result: Purrs4Peace : 1
Result: livemusic : 1
Result: VIPremium : 1
Result: Frome : 1
Result: SundayRoast : 1
Result: Millennials : 1
Result: HealthWithKier : 1
Result: LPs30DaysofGratitude : 1
Result: cooking : 1
Result: gameinsight : 1
Result: Countryfile : 1
Result: androidgames : 1