import java.io.File;
import java.io.IOException;
import java.util.Collection;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Map.Entry;
import org.apache.commons.io.FileUtils;
import backtype.storm.Config;
import backtype.storm.LocalCluster;
import backtype.storm.spout.SpoutOutputCollector;
import backtype.storm.task.OutputCollector;
import backtype.storm.task.TopologyContext;
import backtype.storm.topology.OutputFieldsDeclarer;
import backtype.storm.topology.TopologyBuilder;
import backtype.storm.topology.base.BaseRichBolt;
import backtype.storm.topology.base.BaseRichSpout;
import backtype.storm.tuple.Fields;
import backtype.storm.tuple.Tuple;
import backtype.storm.tuple.Values;
import backtype.storm.utils.Utils;
/**
* 单词计数
* 监控d:\\test目录下面的文件，统计单词出现的总次数
* 当有新文件出现的时候，也要能解析出来
*
* @author Administrator
*
*/
public class LocalTopologyWordCount {
/**
* spout需要继承baserichspout，实现未实现的方法
* @author Administrator
*
*/
public static class DataSourceSpout extends BaseRichSpout{
private Map conf;
private TopologyContext context;
private SpoutOutputCollector collector;
/**
* 初始化方法，只会执行一次
* 在这里面可以写一个初始化的代码
* Map conf：其实里面保存的是topology的一些配置信息
* TopologyContext context：topology的上下文，类似于servletcontext
* SpoutOutputCollector collector：发射器，负责向外发射数据(tuple)
*/
@Override
public void open(Map conf, TopologyContext context,
SpoutOutputCollector collector) {
this.conf = conf;
this.context = context;
this.collector = collector;
}
/**
* 这个方法是spout中最重要的方法，
* 这个方法会被storm框架循环调用，可以理解为这个方法是在一个while循环之内
* 每调用一次，会向外发射一条数据
*/
@Override
public void nextTuple() {
//获取指定目录下面的新文件，
Collection<File> listFiles = FileUtils.listFiles(new File("d:\\test"), new String[]{"txt"}, true);
//分别读取每个文件
for (File file : listFiles) {
try {
List<String> readLines = FileUtils.readLines(file);
for (String line : readLines) {
//把每一行封装成一个tuple，发射出去
this.collector.emit(new Values(line));
}
FileUtils.moveFile(file, new File(file.getAbsolutePath()+System.currentTimeMillis()));//给文件该名字,否则会一直处理这个文件.
} catch (IOException e) {
e.printStackTrace();
}
}
}
/**
* 声明输出字段
*/
@Override
public void declareOutputFields(OutputFieldsDeclarer declarer) {
//给values中的数据起个名字，方便后面的bolt从这个values中取数据
//fields中定义的参数和values中传递的数值是一一对应的
declarer.declare(new Fields("line"));
}
}
/**
* 自定义bolt需要实现baserichbolt
* @author Administrator
*
*/
public static class SplitBolt extends BaseRichBolt{
private Map stormConf;
private TopologyContext context;
private OutputCollector collector;
/**
* 和spout中的open方法意义一样
*/
@Override
public void prepare(Map stormConf, TopologyContext context,
OutputCollector collector) {
this.stormConf = stormConf;
this.context = context;
this.collector = collector;
}
/**
* 是bolt中最重要的方法，当spout发射一个tuple出来，execute也会被调用，需要对spout发射出来的tuple进行处理
*/
@Override
public void execute(Tuple input) {
//获取每一行数据进行切割
String line = input.getStringByField("line");
String[] splits = line.split("\t");
//把切割出来的单词一个一个发射出去
for (String word : splits) {
this.collector.emit(new Values(word));
}
}
//在这没必要定义了，因为execute方法中没有向外发射tuple，所以就不需要声明了。
//如果nextTuple或者execute方法中向外发射了tuple，那么declareOutputFields必须要声明，否则不需要声明
/**
* 声明输出字段
*/
@Override
public void declareOutputFields(OutputFieldsDeclarer declarer) {
declarer.declare(new Fields("word"));
}
}
/**
* 自定义bolt需要实现baserichbolt
* @author Administrator
*
*/
public static class CountBolt extends BaseRichBolt{
private Map stormConf;
private TopologyContext context;
private OutputCollector collector;
/**
* 和spout中的open方法意义一样
*/
@Override
public void prepare(Map stormConf, TopologyContext context,
OutputCollector collector) {
this.stormConf = stormConf;
this.context = context;
this.collector = collector;
}
HashMap<String, Integer> hashMap = new HashMap<String, Integer>();
/**
* 是bolt中最重要的方法，当spout发射一个tuple出来，execute也会被调用，需要对spout发射出来的tuple进行处理
*/
@Override
public void execute(Tuple input) {
//获取每一个单词
String word = input.getStringByField("word");
//在map中进行统计
Integer integer = hashMap.get(word);
if(integer==null){
integer=0;
}
integer++;
hashMap.put(word, integer);
//把这个统计结果打印到控制台
Utils.sleep(1000);
System.out.println("=========================================");
for (Entry<String, Integer> entry : hashMap.entrySet()) {
System.out.println(entry);
}
}
//在这没必要定义了，因为execute方法中没有向外发射tuple，所以就不需要声明了。
//如果nextTuple或者execute方法中向外发射了tuple，那么declareOutputFields必须要声明，否则不需要声明
/**
* 声明输出字段
*/
@Override
public void declareOutputFields(OutputFieldsDeclarer declarer) {
}
}
/**
* 注意：在组装topology的时候，组件的id在定义的时候，名称不能以__开头。__是系统保留的
* @param args
*/
public static void main(String[] args) {
//组装topology
TopologyBuilder topologyBuilder = new TopologyBuilder();
topologyBuilder.setSpout("spout1", new DataSourceSpout());
//.shuffleGrouping("spout1"); 表示让MyBolt接收MySpout发射出来的tuple
topologyBuilder.setBolt("bolt1", new SplitBolt()).shuffleGrouping("spout1");
topologyBuilder.setBolt("bolt2", new CountBolt()).shuffleGrouping("bolt1");
//创建本地storm集群
LocalCluster localCluster = new LocalCluster();
localCluster.submitTopology("wordCountTopology", new Config(), topologyBuilder.createTopology());
}
}