adding world count sample

storm-samples/src/main/java/org/storm/sample/wordcount/WordCountTopology.java

@@ -0,0 +1,58 @@
+/*
+ * Copyright (c) 2015, WSO2 Inc. (http://www.wso2.org) All Rights Reserved.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.storm.sample.wordcount;
+
+import backtype.storm.Config;
+import backtype.storm.LocalCluster;
+import backtype.storm.StormSubmitter;
+import backtype.storm.topology.TopologyBuilder;
+import backtype.storm.tuple.Fields;
+import org.storm.sample.wordcount.bolt.SplitSentence;
+import org.storm.sample.wordcount.bolt.WordCount;
+import org.storm.sample.wordcount.spout.LineStreamer;
+
+/**
+ * This topology demonstrates Storm's stream groupings capabilities using word counting as an example.
+ */
+public class WordCountTopology {
+
+    public static void main(String[] args) throws Exception {
+
+        TopologyBuilder builder = new TopologyBuilder();
+
+        builder.setSpout("line", new LineStreamer(), 5);
+        builder.setBolt("split", new SplitSentence(), 8).shuffleGrouping("line");
+        builder.setBolt("count", new WordCount(), 12).fieldsGrouping("split", new Fields("word"));
+
+        Config conf = new Config();
+        conf.setDebug(false);
+
+
+        if (args != null && args.length > 0) {
+            conf.setNumWorkers(3);
+            StormSubmitter.submitTopology(args[0], conf, builder.createTopology());
+        } else {
+            conf.setMaxTaskParallelism(3);
+            LocalCluster cluster = new LocalCluster();
+            cluster.submitTopology("word-count", conf, builder.createTopology());
+
+            Thread.sleep(10000);
+
+            cluster.shutdown();
+        }
+    }
+}

storm-samples/src/main/java/org/storm/sample/wordcount/bolt/SplitSentence.java

@@ -0,0 +1,66 @@
+/*
+ * Copyright (c) 2015, WSO2 Inc. (http://www.wso2.org) All Rights Reserved.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.storm.sample.wordcount.bolt;
+
+import backtype.storm.task.OutputCollector;
+import backtype.storm.task.TopologyContext;
+import backtype.storm.topology.IRichBolt;
+import backtype.storm.topology.OutputFieldsDeclarer;
+import backtype.storm.tuple.Fields;
+import backtype.storm.tuple.Tuple;
+import backtype.storm.tuple.Values;
+
+import java.util.Map;
+
+public class SplitSentence implements IRichBolt {
+    private OutputCollector collector;
+
+    public SplitSentence() {
+    }
+
+    @Override
+    public void declareOutputFields(OutputFieldsDeclarer declarer) {
+        declarer.declare(new Fields("word"));
+    }
+
+    @Override
+    public Map<String, Object> getComponentConfiguration() {
+        return null;
+    }
+
+    @Override
+    public void prepare(Map stormConf, TopologyContext context, OutputCollector collector) {
+        this.collector = collector;
+    }
+
+    @Override
+    public void execute(Tuple input) {
+        String line = input.getStringByField("line");
+        line = line.replaceAll("\\.", "");
+        if (line != null) {
+            String[] tokens = line.split("\\s");
+            for (String token : tokens) {
+                collector.emit(new Values(token));
+            }
+        }
+    }
+
+    @Override
+    public void cleanup() {
+
+    }
+}

storm-samples/src/main/java/org/storm/sample/wordcount/bolt/WordCount.java

@@ -0,0 +1,53 @@
+/*
+ * Copyright (c) 2015, WSO2 Inc. (http://www.wso2.org) All Rights Reserved.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.storm.sample.wordcount.bolt;
+
+import backtype.storm.topology.BasicOutputCollector;
+import backtype.storm.topology.OutputFieldsDeclarer;
+import backtype.storm.topology.base.BaseBasicBolt;
+import backtype.storm.tuple.Fields;
+import backtype.storm.tuple.Tuple;
+import backtype.storm.tuple.Values;
+
+import java.util.HashMap;
+import java.util.Map;
+
+/**
+* Created on 6/27/15.
+*/
+public class WordCount extends BaseBasicBolt {
+    Map<String, Integer> counts = new HashMap<String, Integer>();
+
+    @Override
+    public void execute(Tuple tuple, BasicOutputCollector collector) {
+
+        String word = tuple.getString(0);
+        Integer count = counts.get(word);
+        if (count == null) {
+            count = 0;
+        }
+        count++;
+        counts.put(word, count);
+        System.out.println(word +"=>"+count);
+        collector.emit(new Values(word, count));
+    }
+
+    @Override
+    public void declareOutputFields(OutputFieldsDeclarer declarer) {
+        declarer.declare(new Fields("word", "count"));
+    }
+}