test_class.file: storm/trident-examples/src/test/java/storm/trident/topology/SentimentAnalysisTest.java
test_class.identifier: SentimentAnalysisTest
test_case.body: @Test
      public void testRunTopology() throws InterruptedException {
          Configuration conf = new Configuration();
          SentimentAnalysis topology = new SentimentAnalysis("sa", conf);
          
          StormRunner.runTopologyLocally(topology.buildTopology(), "sa", conf, 100);
      }
test_case.class_method_signature: SentimentAnalysisTest.testRunTopology()
test_case.full_signature: @Test public void testRunTopology()
test_case.identifier: testRunTopology
test_case.invocations: runTopologyLocally buildTopology
test_case.modifiers: @Test public
test_case.parameters: ()
test_case.return: void
test_case.signature: void testRunTopology()
focal_class.declarator: spoutType
focal_class.modifier: private
focal_class.original_string: private String spoutType;
focal_class.type: String
focal_class.var_name: spoutType
focal_class.declarator: sinkType
focal_class.modifier: private
focal_class.original_string: private String sinkType;
focal_class.type: String
focal_class.var_name: sinkType
focal_class.declarator: spoutThreads
focal_class.modifier: private
focal_class.original_string: private int spoutThreads;
focal_class.type: int
focal_class.var_name: spoutThreads
focal_class.declarator: parserThreads
focal_class.modifier: private
focal_class.original_string: private int parserThreads;
focal_class.type: int
focal_class.var_name: parserThreads
focal_class.declarator: tweetFilterThreads
focal_class.modifier: private
focal_class.original_string: private int tweetFilterThreads;
focal_class.type: int
focal_class.var_name: tweetFilterThreads
focal_class.declarator: textFilterThreads
focal_class.modifier: private
focal_class.original_string: private int textFilterThreads;
focal_class.type: int
focal_class.var_name: textFilterThreads
focal_class.declarator: stemmerThreads
focal_class.modifier: private
focal_class.original_string: private int stemmerThreads;
focal_class.type: int
focal_class.var_name: stemmerThreads
focal_class.declarator: posScorerThreads
focal_class.modifier: private
focal_class.original_string: private int posScorerThreads;
focal_class.type: int
focal_class.var_name: posScorerThreads
focal_class.declarator: negScorerThreads
focal_class.modifier: private
focal_class.original_string: private int negScorerThreads;
focal_class.type: int
focal_class.var_name: negScorerThreads
focal_class.declarator: scorerThreads
focal_class.modifier: private
focal_class.original_string: private int scorerThreads;
focal_class.type: int
focal_class.var_name: scorerThreads
focal_class.declarator: sinkThreads
focal_class.modifier: private
focal_class.original_string: private int sinkThreads;
focal_class.type: int
focal_class.var_name: sinkThreads
focal_class.file: storm/trident-examples/src/main/java/storm/trident/topology/SentimentAnalysis.java
focal_class.identifier: SentimentAnalysis
focal_class.class_method_signature: SentimentAnalysis.SentimentAnalysis(String topologyName, Configuration config)
focal_class.full_signature: public  SentimentAnalysis(String topologyName, Configuration config)
focal_class.identifier: SentimentAnalysis
focal_class.modifiers: public
focal_class.parameters: (String topologyName, Configuration config)
focal_class.signature:  SentimentAnalysis(String topologyName, Configuration config)
focal_class.class_method_signature: SentimentAnalysis.initialize()
focal_class.full_signature: @Override public void initialize()
focal_class.identifier: initialize
focal_class.modifiers: @Override public
focal_class.parameters: ()
focal_class.return: void
focal_class.signature: void initialize()
focal_class.class_method_signature: SentimentAnalysis.buildTopology()
focal_class.full_signature: @Override public StormTopology buildTopology()
focal_class.identifier: buildTopology
focal_class.modifiers: @Override public
focal_class.parameters: ()
focal_class.return: StormTopology
focal_class.signature: StormTopology buildTopology()
focal_class.superclass: extends AbstractTopology
focal_method.body: @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();
      }
focal_method.class_method_signature: SentimentAnalysis.buildTopology()
focal_method.full_signature: @Override public StormTopology buildTopology()
focal_method.identifier: buildTopology
focal_method.invocations: getInt getString newStream createKafkaSpout shuffle parallelismHint project shuffle parallelismHint each project shuffle parallelismHint each project shuffle parallelismHint each project shuffle parallelismHint each project shuffle parallelismHint each project shuffle parallelismHint each parallelismHint join shuffle parallelismHint each equals parallelismHint each getOutputFields build
focal_method.modifiers: @Override public
focal_method.parameters: ()
focal_method.return: StormTopology
focal_method.signature: StormTopology buildTopology()