Kafka实战－数据持久化 - 哥不是小萝莉

package cn.hadoop.hdfs.storm;

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

import redis.clients.jedis.Jedis;
import cn.hadoop.hdfs.util.JedisFactory;
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;

/**
* @Date Jun 10, 2015
*
* @Author dengjie
*
* @Note Calc WordsCount eg.
*/
public class WordsCounterBlots implements IRichBolt {

/**
*
*/
private static final long serialVersionUID = -619395076356762569L;

OutputCollector collector;
Map<String, Integer> counter;

@SuppressWarnings("rawtypes")
public void prepare(Map stormConf, TopologyContext context, OutputCollector collector) {
this.collector = collector;
this.counter = new HashMap<String, Integer>();
}

public void execute(Tuple input) {
String word = input.getString(0);
Integer integer = this.counter.get(word);
if (integer != null) {
integer += 1;
this.counter.put(word, integer);
} else {
this.counter.put(word, 1);
}
for (Entry<String, Integer> entry : this.counter.entrySet()) {
// write result to redis
Jedis jedis = JedisFactory.getJedisInstance("real-time");
jedis.set(entry.getKey(), entry.getValue().toString());

// write result to mysql
// ...
}
this.collector.ack(input);
}

public void cleanup() {
// TODO Auto-generated method stub

}

public void declareOutputFields(OutputFieldsDeclarer declarer) {
// TODO Auto-generated method stub

}

public Map<String, Object> getComponentConfiguration() {
// TODO Auto-generated method stub
return null;
}

}