TrendingTopics extends AbstractTopology { @Override public StormTopology buildTopology() { TridentTopology topology = new TridentTopology(); Stream rawTweets = null; switch (spoutType) { case BaseConstants.SourceType.FILE: int maxBatchSize = config.getInt(Config.SPOUT_MAX_BATCH_SIZE, 100); String spoutPath = config.getString(Config.SPOUT_PATH); rawTweets = topology.newStream("spout", new FileSpout(maxBatchSize, spoutPath)); break; case BaseConstants.SourceType.KAFKA: createKafkaSpout(spoutType, spoutType); break; default: throw new IllegalArgumentException("Must inform a spout type"); } rawTweets = rawTweets.parallelismHint(spoutThreads) .shuffle(); Stream tweets = rawTweets.each(new Fields("tweet_str"), new JsonParser(), new Fields("tweet")) .parallelismHint(parserThreads) .shuffle() .project(new Fields("tweet")); GroupedStream topics = tweets.each(new Fields("tweet"), new TopicExtractor(), new Fields("topic")) .parallelismHint(topicExtractorThreads) .groupBy(new Fields("topic")); Stream counts = topics.persistentAggregate(new MemoryMapState.Factory(), new Count(), new Fields("count")) .parallelismHint(counterThreads) .newValuesStream(); Stream trending = counts.persistentAggregate(new MemoryMapState.Factory(), new Fields("topic", "count"), new MaxCount(), new Fields("top")) .newValuesStream() .parallelismHint(trendingThreads); if (sinkType.equals(SinkType.CONSOLE)) { trending.each(new Fields("top"), new Debug()) .parallelismHint(sinkThreads); } return topology.build(); } TrendingTopics(String topologyName, Configuration config); @Override void initialize(); @Override StormTopology buildTopology();  }
@Test public void testRunTopology() throws InterruptedException { Configuration conf = new Configuration(); TrendingTopics topology = new TrendingTopics("tt", conf); StormRunner.runTopologyLocally(topology.buildTopology(), "tt", conf, 100); }