二、Storm入门之Hello Storm（续）

package bolts;

import java.util.ArrayList;
import java.util.List;
import java.util.Map;

import backtype.storm.task.OutputCollector;
import backtype.storm.task.TopologyContext;
import backtype.storm.topology.IRichBolt;
import backtype.storm.topology.OutputFieldsDeclarer;
import backtype.storm.tuple.Fields;
import backtype.storm.tuple.Tuple;
import backtype.storm.tuple.Values;

public class WordNormalizer implements IRichBolt{
private OutputCollector collector;

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"));
}

}

package bolts;

import java.util.HashMap;
import java.util.Map;

import backtype.storm.task.OutputCollector;
import backtype.storm.task.TopologyContext;
import backtype.storm.topology.IRichBolt;
import backtype.storm.topology.OutputFieldsDeclarer;
import backtype.storm.tuple.Tuple;

public class WordCounter implements IRichBolt{
Integer id;
String name;
Map<String,Integer>counters;

private OutputCollector collector;

/**
* At the end of the spout (when the cluster is shutdown
* We will show the word counters
*/

@Override

public void cleanup(){
System.out.println("-- Word Counter ["+name+"-"+id+"]--");
for(Map.Entry<String,Integer>entry: counters.entrySet()){
System.out.println(entry.getKey()+": "+entry.getValue());
}
}

/**
* On each word We will count
*/
@Override

public void execute(Tuple input) {
String str =input.getString(0);
/**
* If the word dosn't exist in the map we will create
* this, if not We will add 1
*/
if(!counters.containsKey(str)){
counters.put(str,1);
}else{
Integer c =counters.get(str) +1;
counters.put(str,c);
}
//Set the tuple as Acknowledge
collector.ack(input);
}

/**
* On create
*/

@Override

public void prepare(Map stormConf,TopologyContext context,OutputCollector collector) {
this.counters=newHashMap<String,Integer>();
this.collector=collector;
this.name=context.getThisComponentId();
this.id=context.getThisTaskId();
}

@Override

public void declareOutputFields(OutputFieldsDeclarer declarer) {}

}

主类

TopologyBuilder builder =new TopologyBuilder();
builder.setSpout("word-reader",new WordReader());
builder.setBolt("word-normalizer",new WordNormalizer()).shuffleGrouping("word-reader");
builder.setBolt("word-counter",new WordCounter(),2).fieldsGrouping("word-normalizer",new Fields("word"));

Config conf =new Config();
conf.put("wordsFile",args[0]);
conf.setDebug(false);

LocalCluster cluster =new LocalCluster();
cluster.submitTopology("Getting-Started-Toplogie",conf,builder.createTopology());
Thread.sleep(1000);
cluster.shutdown();

import spouts.WordReader;
import bolts.WordCounter;
import bolts.WordNormalizer;

import backtype.storm.Config;
import backtype.storm.LocalCluster;
import backtype.storm.topology.TopologyBuilder;
import backtype.storm.tuple.Fields;

public class TopologyMain{
public static void main(String[]args)throws InterruptedException{
//Topology definition
TopologyBuilder builder =new TopologyBuilder();
builder.setSpout("word-reader",new WordReader());
builder.setBolt("word-normalizer",new WordNormalizer()).shuffleGrouping("word-reader");
builder.setBolt("word-counter",new WordCounter(),2).fieldsGrouping("word-normalizer",new Fields("word"));

//Configuration
Config conf =new Config();
conf.put("wordsFile",args[0]);
conf.setDebug(false);

//Topology run
conf.put(Config.TOPOLOGY_MAX_SPOUT_PENDING,1);
LocalCluster cluster =new LocalCluster();
cluster.submitTopology("Getting-Started-Toplogie",conf,builder.createTopology());
Thread.sleep(1000);
cluster.shutdown();
}

}

运行本项目

builder.setBolt(

"word-counter"

,

new

WordCounter(),

2

).shuffleGrouping(

"word-normalizer"

);

总结