Apache Storm在Twitter上的应用

由 chalex 创建，小路依依 最后一次修改 2016-12-12 在本章中，我们将讨论Apache Storm的实时应用程序。我们将看到Storm如何在Twitter中使用。Twitter Twitter是一种在线社交网络服务，提供发送和接收用户推文的平台。注册用户可以阅读和发布tweet，但未注册的用户只能阅读tweets。 Hashtag用于按关键字在相关关键字之前附加＃来对tweet进行分类。现在让我们来看一个实时场景，找到每个主题使用最多的hashtag。Spout创建spout的目的是尽快收到人们提交的tweets。Twitter提供了“Twitter Streaming API”，一个基于Web服务的工具，用于实时检索人们提交的tweets。Twitter Streaming API可以使用任何编程语言访问。 twitter4j是一个开源的非官方Java库，它提供了一个基于Java的模块，可以轻松访问Twitter Streaming API。twitter4j提供了一个基于监听器的框架来访问tweet。要访问Twitter Streaming API，我们需要登录Twitter开发人员帐户，并获取以下OAuth身份验证详细信息。 Customerkey CustomerSecret 的accessToken AccessTookenSecret Storm在其入门套件中提供了一个twitter spout，TwitterSampleSpout。我们将使用它来检索tweet。该邮件需要OAuth身份验证详细信息和至少一个关键字。该spout将发出基于关键字的实时tweet。完整的程序代码如下。编码：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")); }} Hashtag阅读器spout由spout发出的tweet将被转发到HashtagReaderBolt，它将处理该tweet并发出所有可用的hashtag。HashtagReaderBolt使用twitter4j提供的getHashTagEntities方法。getHashTagEntities读取tweet并返回hashtag的列表。完整的程序代码如下 - 编码：HashtagReaderBolt.java import java.util.HashMap;import java.util.Map;import twitter4j.*;import twitter4j.co