Getting Started - Getting Started with Storm - page 14

public void cleanup () {}
/**
* The bolt will receive the line from the
* words file and process it to Normalize this line
*
* The normalize will be put the words in lower case
* and split the line to get all words in this
*/
public void execute ( Tuple input ) {
String sentence = input . getString ( 0 );
String [] words = sentence . split ( " " );
for ( String word : words ){
word = word . trim ();
if (! word . isEmpty ()){
word = word . toLowerCase ();
//Emit the word
List a = new ArrayList ();
a . add ( input );
collector . emit ( a , new Values ( word ));
}
}
// Acknowledge the tuple
collector . ack ( input );
}
public void prepare ( Map stormConf , TopologyContext context ,
OutputCollector collector ) {
this . collector = collector ;
}
/**
* The bolt will only emit the field "word"
*/
public void declareOutputFields ( OutputFieldsDeclarer declarer ) {
declarer . declare ( new Fields ( "word" ));
}
}
In this class, we see an example of emitting multiple tuples in a single
execute call. If the method receives the sentence This is the Storm
topic , in a single execute call, it will emit five new tuples.
The next bolt, WordCounter , will be responsible for counting words. When the topology
finishes (when the cleanup() method is called), we'll show the count for each word.
This is an example of a bolt that emits nothing. In this case, the data is
added to a map, but in real life the bolt could store data to a database.