Merge pull request #7 from anapetrova/kafka

Kafka integration into Storm + POM file for installation

Author: Davis Liang

scripts/startKafkaServer.sh

@@ -0,0 +1,7 @@
+#!/bin/bash
+echo "starting kafka client"
+
+KAFKAHOME="$HOME/kafka-0.10.1.1-src"
+
+# Added the sudo due to file permissions being messed up
+sudo $KAFKAHOME/bin/kafka-server-start.sh $KAFKAHOME/config/server.properties

scripts/startNimbus.sh

@@ -1,2 +1,2 @@
 #!/bin/bash
-sudo ~/apache-storm-0.9.5/bin/storm nimbus
+sudo ~/apache-storm-1.0.3/bin/storm nimbus

scripts/startStormUI.sh

@@ -1,2 +1,2 @@
 #!/bin/bash
-sudo ~/apache-storm-0.9.5/bin/storm ui
+sudo ~/apache-storm-1.0.3/bin/storm ui

scripts/startSupervisor.sh

@@ -1,2 +1,2 @@
 #!/bin/bash
-sudo ~/apache-storm-0.9.5/bin/storm supervisor
+sudo ~/apache-storm-1.0.3/bin/storm supervisor

scripts/systemStartMac.sh

@@ -8,3 +8,5 @@ sleep 5
 ~/tweetnet/scripts/newTerminalMac.sh ~/tweetnet/scripts/startSupervisor.sh
 sleep 5
 ~/tweetnet/scripts/newTerminalMac.sh ~/tweetnet/scripts/startStormUI.sh
+sleep 6
+~/tweetnet/scripts/newTerminalMac.sh ~/tweetnet/scripts/startKafkaServer.sh

scripts/systemStartUbuntu.sh

@@ -8,3 +8,5 @@ sleep 5
 gnome-terminal -e ~/tweetnet/scripts/startSupervisor.sh
 sleep 5
 gnome-terminal -e ~/tweetnet/scripts/startStormUI.sh
+sleep 5
+gnome-terminal -e ~/tweetnet/scripts/startKafkaServer.sh

src/storm/TwitterCleanerBolt.java

@@ -5,14 +5,14 @@
 import twitter4j.*;
 import twitter4j.conf.*;
 import java.util.List;
-import backtype.storm.tuple.Fields;
-import backtype.storm.tuple.Values;
+import org.apache.storm.tuple.Fields;
+import org.apache.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;
+import org.apache.storm.task.OutputCollector;
+import org.apache.storm.task.TopologyContext;
+import org.apache.storm.topology.IRichBolt;
+import org.apache.storm.topology.OutputFieldsDeclarer;
+import org.apache.storm.tuple.Tuple;
 
 import java.io.FileOutputStream;
 import java.io.OutputStream;
@@ -29,7 +29,7 @@ public class TwitterCleanerBolt implements IRichBolt {
  boolean useTopicSelector = false;
  String language = new String("en");
  List<String> topics = Arrays.asList("politics","entertainment",
- "world","us","business","opinion","tech","science","health", 
+ "world","us","business","opinion","tech","science","health",
  "sports", "art", "style", "food", "travel");
 
  /**
@@ -56,7 +56,7 @@ public void prepare(Map conf, TopologyContext context, OutputCollector collector
  public void execute(Tuple tuple) {
 
  Status tweet = (Status) tuple.getValueByField("tweet");
- 
+
  // return criterion
  if(!tweet.getLang().equals(language))
  return;
@@ -71,15 +71,13 @@ public void execute(Tuple tuple) {
  txt = this.removeUrl(txt);
  txt = txt.replace("\n", "");
  txt = txt.toLowerCase();
- 
- 
+
+
  // extract hashtags
  String hasht = "\nhashtags: ";
  boolean keep = false;
  for(HashtagEntity hashtage : tweet.getHashtagEntities()) {
- this.collector.emit(new Values(hashtage.getText())); 
-
- // only select tweets that have hashtags/topics co-occurrence. 
+ // only select tweets that have hashtags/topics co-occurrence.
  if(this.useTopicSelector == true){
 
  for(String s:this.topics) {
@@ -100,15 +98,17 @@ public void execute(Tuple tuple) {
 
  //removes multiple whitespace, hashtag entries, and tag entries
  String finaltext = txt.replaceAll("#[^\\s]+","").replaceAll("@[^\\s]+","").replaceAll("( )+", " ");
- 
+
  //remove characters we don't want
- finaltext = preserveASCII(finaltext); 
- 
+ finaltext = preserveASCII(finaltext);
+
+ //emit onto the kafka bolt
+ this.collector.emit(new Values(finaltext + hasht));
+
  finaltext = "\n\ntext: " + finaltext;
- 
+
  if(finaltext.length()<60)
  return;
-
  try {
  oStream = new FileOutputStream(System.getProperty("user.home")+"/tweetnet/data/dump.txt", true);
  oStream.write(finaltext.getBytes());
@@ -118,7 +118,6 @@ public void execute(Tuple tuple) {
  // TODO Auto-generated catch block
  e.printStackTrace();
  }
- 
  }
 
  /**
@@ -134,7 +133,7 @@ public void cleanup() {
  **/
  @Override
  public void declareOutputFields(OutputFieldsDeclarer declarer) {
- declarer.declare(new Fields("postCleanedTweets"));
+ declarer.declare(new Fields("message"));
  }
 
  /**
@@ -181,10 +180,10 @@ public static char[] removeChar( char[] original, int removeLocation) {
  public static String removeUrl(String tweet) {
  try{
  String urlPattern = "((https?|ftp|gopher|telnet|file|Unsure|http|https):((//)|(\\\\))+[\\w\\d:#@%/;$()~_?\\+-=\\\\\\.&]*)";
- 
+
  Pattern p = Pattern.compile(urlPattern,Pattern.CASE_INSENSITIVE);
  Matcher m = p.matcher(tweet);
- 
+
  // re-assigns str while removing URLs
  int i = 0;
  while (m.find()) {
@@ -199,4 +198,4 @@ public static String removeUrl(String tweet) {
  }
  }
 
-} 
+}

src/storm/TwitterStorm.java

@@ -1,14 +1,18 @@
 
 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;
-
+import org.apache.storm.tuple.Fields;
+import org.apache.storm.tuple.Values;
+import org.apache.storm.Config;
+import org.apache.storm.LocalCluster;
+import org.apache.storm.topology.TopologyBuilder;
+import org.apache.storm.kafka.bolt.KafkaBolt;
+import org.apache.storm.kafka.bolt.selector.DefaultTopicSelector;
+import org.apache.storm.kafka.bolt.mapper.*;
 /**
- * Main class for storm topology. 
+ * Main class for storm topology.
  */
+
+
 public class TwitterStorm {
 
  /**
@@ -29,19 +33,36 @@ public static void main(String[] args) throws Exception{
  String[] arguments = args.clone();
  String[] keyWords = Arrays.copyOfRange(arguments, 4, arguments.length);
 
- //create a new Storm configuration. 
+ //create a new Storm configuration.
  Config config = new Config();
  config.setDebug(true);
 
- //create a new topology. 
+ //create a new topology.
  TopologyBuilder builder = new TopologyBuilder();
 
- builder.setSpout("streamSpout", new TwitterStreamSpout(
- consumerKey,consumerSecret, accessToken, accessTokenSecret, keyWords));
+ TwitterStreamSpout streamSpout = new TwitterStreamSpout(
+ consumerKey,consumerSecret, accessToken, accessTokenSecret, keyWords);
+
+ // streamSpout.scheme = new SchemeAsMultiScheme(new KafkaBoltKeyValueScheme());
+ builder.setSpout("streamSpout", streamSpout);
+
+ TwitterCleanerBolt cleanerBolt = new TwitterCleanerBolt();
+
+ builder.setBolt("cleanerBolt", cleanerBolt).shuffleGrouping("streamSpout");
+
+ Properties props = new Properties();
+ props.put("bootstrap.servers", "localhost:9092");
+ props.put("acks", "1");
+ props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
+ props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");
+
+ KafkaBolt kafkaBolt = new KafkaBolt()
+ .withProducerProperties(props)
+ .withTopicSelector(new DefaultTopicSelector("twitterstorm"))
+ .withTupleToKafkaMapper(new FieldNameBasedTupleToKafkaMapper());
+
+ builder.setBolt("forwardToKafka", kafkaBolt).shuffleGrouping("cleanerBolt");
 
- builder.setBolt("cleanerBolt", new TwitterCleanerBolt())
- .shuffleGrouping("streamSpout");
- 
  //submit topology to local cluster.
  LocalCluster cluster = new LocalCluster();
  cluster.submitTopology("TwitterHashtagStorm", config,

src/storm/TwitterStreamSpout.java

@@ -1,4 +1,4 @@
- 
+
 import java.util.Map;
 import java.util.concurrent.LinkedBlockingQueue;
 
@@ -13,16 +13,16 @@
 import twitter4j.auth.AccessToken;
 import twitter4j.conf.ConfigurationBuilder;
 
-import backtype.storm.Config;
-import backtype.storm.spout.SpoutOutputCollector;
+import org.apache.storm.Config;
+import org.apache.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 org.apache.storm.task.TopologyContext;
+import org.apache.storm.topology.OutputFieldsDeclarer;
+import org.apache.storm.topology.base.BaseRichSpout;
+import org.apache.storm.tuple.Fields;
+import org.apache.storm.tuple.Values;
 
-import backtype.storm.utils.Utils;
+import org.apache.storm.utils.Utils;
 
 /**
  * this class talks directly the the twitterAPI using the user credentials
@@ -34,13 +34,13 @@ public class TwitterStreamSpout extends BaseRichSpout {
  SpoutOutputCollector _collector;
  LinkedBlockingQueue<Status> queue = null;
  TwitterStream _twitterStream;
- 
+
  String consumerKey;
  String consumerSecret;
  String accessToken;
  String accessTokenSecret;
  String[] keyWords;
- 
+
  /**
  * Constructor.
  * @param consumerKey Twitter API credential
@@ -58,14 +58,14 @@ public TwitterStreamSpout(String consumerKey, String consumerSecret,
  this.accessTokenSecret = accessTokenSecret;
  this.keyWords = keyWords;
  }
- 
+
  /**
  * TO DO: default constructor is a stub.
  */
  public TwitterStreamSpout() {
  // TODO Auto-generated constructor stub
  }
- 
+
  /**
  * creates a new status blockingQueue and a statusListener.
  * @param conf Storm configuration
@@ -85,66 +85,66 @@ public void open(Map conf, TopologyContext context,
  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(); 
- 
+
+ 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);
  }
  }
- 
+
  /**
  * polls from the blocking queue to get next status.
  */
  @Override
  public void nextTuple() {
  Status ret = queue.poll();
- 
+
  if (ret == null) {
  Utils.sleep(50);
  } else {
  _collector.emit(new Values(ret));
  }
  }
- 
+
  /**
- * closes twitter stream. 
+ * closes twitter stream.
  */
  @Override
  public void close() {
  _twitterStream.shutdown();
  }
- 
+
  /**
  * worker node configurator. Default set to 1 local machine.
  */
@@ -154,13 +154,13 @@ public Map<String, Object> getComponentConfiguration() {
  ret.setMaxTaskParallelism(1);
  return ret;
  }
- 
+
  @Override
  public void ack(Object id) {}
- 
+
  @Override
  public void fail(Object id) {}
- 
+
  /**
  * Declare output field type.
  */