WordCount extends AbstractTopology { @Override public StormTopology buildTopology() { TridentTopology topology = new TridentTopology(); Stream sentences = null; switch (spoutType) { case BaseConstants.SourceType.FILE: int maxBatchSize = config.getInt(Config.SPOUT_MAX_BATCH_SIZE, 100); String spoutPath = config.getString(Config.SPOUT_PATH); sentences = 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"); } sentences = sentences.parallelismHint(spoutThreads); GroupedStream words = sentences.each(new Fields("sentence"), new Split(), new Fields("word")) .parallelismHint(splitterThreads) .groupBy(new Fields("word")); Stream counts = words.persistentAggregate(new MemoryMapState.Factory(), new Count(), new Fields("count")) .parallelismHint(counterThreads) .newValuesStream(); if (sinkType.equals(SinkType.CONSOLE)) { counts.each(new Fields("word", "count"), new Debug()) .parallelismHint(sinkThreads); } return topology.build(); } WordCount(String topologyName, Configuration config); @Override void initialize(); @Override StormTopology buildTopology();  }
@Test public void testRunTopology() throws InterruptedException { Configuration conf = new Configuration(); WordCount topology = new WordCount("wordcount", conf); StormRunner.runTopologyLocally(topology.buildTopology(), "wordcount", conf, 100); }