10. Example: RollingTopWords Topology

The RollingTopWords.java is included with storm-starter.

package storm.starter;

import backtype.storm.Config;
import backtype.storm.testing.TestWordSpout;
import backtype.storm.topology.TopologyBuilder;
import backtype.storm.tuple.Fields;
import storm.starter.bolt.IntermediateRankingsBolt;
import storm.starter.bolt.RollingCountBolt;
import storm.starter.bolt.TotalRankingsBolt;
import storm.starter.util.StormRunner;

/**
* This topology does a continuous computation of the top N words that the topology has seen in terms of cardinality.
* The top N computation is done in a completely scalable way, and a similar approach could be used to compute things
* like trending topics or trending images on Twitter.
*/
public class RollingTopWords {

 private static final int DEFAULT_RUNTIME_IN_SECONDS = 60;
 private static final int TOP_N = 5;

 private final TopologyBuilder builder;
 private final String topologyName;
 private final Config topologyConfig;
 private final int runtimeInSeconds;

 public RollingTopWords() throws InterruptedException {
 builder = new TopologyBuilder();
 topologyName = "slidingWindowCounts";
 topologyConfig = createTopologyConfiguration();
 runtimeInSeconds = DEFAULT_RUNTIME_IN_SECONDS;

 wireTopology();
 }

 private static Config createTopologyConfiguration() {
 Config conf = new Config();
 conf.setDebug(true);
 return conf;
 }

 private void wireTopology() throws InterruptedException {
 String spoutId = "wordGenerator";
 String counterId = "counter";
 String intermediateRankerId = "intermediateRanker";
 String totalRankerId = "finalRanker";
 builder.setSpout(spoutId, new TestWordSpout(), 5);
 builder.setBolt(counterId, new RollingCountBolt(9, 3), 4).fieldsGrouping(spoutId, new Fields("word"));
 builder.setBolt(intermediateRankerId, new IntermediateRankingsBolt(TOP_N), 4).fieldsGrouping(counterId, new Fields( "obj"));
 builder.setBolt(totalRankerId, new TotalRankingsBolt(TOP_N)).globalGrouping(intermediateRankerId);
 }

 public void run() throws InterruptedException {
 StormRunner.runTopologyLocally(builder.createTopology(), topologyName, topologyConfig, runtimeInSeconds);
 }

 public static void main(String[] args) throws Exception {
 new RollingTopWords().run();
 }
}