SentimentAnalysis extends AbstractTopology { @Override public StormTopology buildTopology() { TridentTopology topology = new TridentTopology(); Stream rawTweets = null; switch (spoutType) { case 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, "tweet_str")); break; case SourceType.KAFKA: createKafkaSpout(spoutType, spoutType); break; default: throw new IllegalArgumentException("Must inform a spout type"); } rawTweets.parallelismHint(spoutThreads) .shuffle(); Stream tweets = rawTweets.each(new Fields("tweet_str"), new JsonParser(), new Fields("tweet")) .parallelismHint(parserThreads) .shuffle() .project(new Fields("tweet")); Stream parsedTweets = tweets.each(new Fields("tweet"), new TweetFilter(), new Fields("id", "text")) .parallelismHint(tweetFilterThreads) .shuffle() .project(new Fields("id", "text")); Stream cleanTweets = parsedTweets.each(new Fields("text"), new TextFilter(), new Fields("text_clean")) .parallelismHint(textFilterThreads) .shuffle() .project(new Fields("id", "text_clean")); Stream stemmedTweets = cleanTweets.each(new Fields("text_clean"), new Stemmer(), new Fields("text_stemmed")) .parallelismHint(stemmerThreads) .shuffle() .project(new Fields("id", "text_stemmed")); Stream posScores = stemmedTweets.each(new Fields("text_stemmed"), new SentimentTypeScorer(Type.Positive), new Fields("pos_score")) .parallelismHint(posScorerThreads) .shuffle() .project(new Fields("id", "text_stemmed", "pos_score")); Stream negScores = stemmedTweets.each(new Fields("text_stemmed"), new SentimentTypeScorer(Type.Negative), new Fields("neg_score")) .parallelismHint(negScorerThreads) .shuffle() .project(new Fields("id", "neg_score")); Stream joinedScores = topology.join(posScores, new Fields("id"), negScores, new Fields("id"), new Fields("id", "text", "pos_score", "neg_score")) .parallelismHint(sinkThreads); Stream scores = joinedScores.each(new Fields("pos_score", "neg_score"), new SentimentScorer(), new Fields("score")) .parallelismHint(scorerThreads) .shuffle(); if (sinkType.equals(SinkType.CONSOLE)) { scores.each(scores.getOutputFields(), new Debug()) .parallelismHint(sinkThreads); } return topology.build(); } SentimentAnalysis(String topologyName, Configuration config); @Override void initialize(); @Override StormTopology buildTopology();  }
@Test public void testRunTopology() throws InterruptedException { Configuration conf = new Configuration(); SentimentAnalysis topology = new SentimentAnalysis("sa", conf); StormRunner.runTopologyLocally(topology.buildTopology(), "sa", conf, 100); }