词频统计

需求：

读取指定目录的数据，并实现单词计数功能

实现方案：

  1. Spout来读取指定目录的数据，作为后续Bolt处理的input
  2. 使用一个Bolt把input的数据，切割开，我们按照逗号进行分割
  3. 使用一个Bolt来进行最终的单词次数统计操作
  4. 输出

涉及到文件操作，为了简单使用commons-io

  1. <dependency>
  2.   <groupId>commons-io</groupId>
  3.   <artifactId>commons-io</artifactId>
  4.   <version>2.4</version>
  5. </dependency>

在E:\stormtest 创建文件test.txt，作为我们的测试数据

内容为：

a,b,c,d
a,b,c
a,c
a,d
f,b

拓扑设计：DataSourceSpout 》SplitBolt》CountBolt

1.定义数据源DataSourceSpout

创建DataSourceSpout继承BaseRichSpout重写open、nextTuple、declareOutputFields

  1.     public static class DataSourceSpout extends BaseRichSpout{
  2.         private SpoutOutputCollector spoutOutputCollector;
  4.         @Override
  5.         public void open(Map map, TopologyContext topologyContext, SpoutOutputCollector spoutOutputCollector) {
  6.             this.spoutOutputCollector = spoutOutputCollector;
  7.         }
  9.         @Override
  10.         public void nextTuple() {
  11.             Collection<File>  files =  FileUtils.listFiles(new File("E:\\stormtest"),new String[]{"txt"},true);
  12.             for (File file:files){
  13.                 try {
  14.                     List<String> lines  =   FileUtils.readLines(file);
  15.                     for (String words:lines){
  16.                         spoutOutputCollector.emit(new Values(words));
  17.                     }
  18.                     //此方法一直循环(nextTuple)，所以要把处理完的文件改名
  19.                     FileUtils.moveFile(file,new File(file.getAbsolutePath()+System.currentTimeMillis()));
  20.                 } catch (IOException e) {
  21.                     e.printStackTrace();
  22.                 }
  23.             }
  25.         }
  27.         @Override
  28.         public void declareOutputFields(OutputFieldsDeclarer outputFieldsDeclarer) {
  29.             outputFieldsDeclarer.declare(new Fields("lines"));
  30.         }
  31.     }

2.定义SplitBolt

创建SplitBolt继承BaseRichBolt，重写prepare、execute、declareOutputFields

  1.     public static class SplitBolt extends BaseRichBolt{
  3.         private OutputCollector outputCollector;
  5.         @Override
  6.         public void prepare(Map map, TopologyContext topologyContext, OutputCollector outputCollector) {
  7.             this.outputCollector = outputCollector;
  8.         }
  10.         @Override
  11.         public void execute(Tuple tuple) {
  12.             String line = tuple.getStringByField("lines");
  13.             String[] words = line.split(",");
  14.             for (String word : words){
  15.                 outputCollector.emit(new Values(word));
  16.             }
  17.         }
  19.         @Override
  20.         public void declareOutputFields(OutputFieldsDeclarer outputFieldsDeclarer) {
  21.             outputFieldsDeclarer.declare(new Fields("word"));
  22.         }
  23.     }

3.定义CountBolt

创建CountBolt继承BaseRichBolt，重写prepare、execute、declareOutputFields

  1. public static class CountBolt extends BaseRichBolt{
  3.         @Override
  4.         public void prepare(Map map, TopologyContext topologyContext, OutputCollector outputCollector) {
  6.         }
  8.         private  Map<String,Integer> map = new HashMap<>();
  11.         @Override
  12.         public void execute(Tuple tuple) {
  13.            String word =  tuple.getStringByField("word");
  14.            Integer count = map.get(word);
  15.            if (null==count){
  16.                count=0;
  17.            }
  18.            count++;
  19.            map.put(word,count);
  20.             System.out.println("=====");
  21.            Set<Map.Entry<String,Integer>>  entrySet = map.entrySet();
  22.            for (Map.Entry<String,Integer> entry:entrySet){
  23.                System.out.println(entry);
  24.            }
  26.         }
  28.         @Override
  29.         public void declareOutputFields(OutputFieldsDeclarer outputFieldsDeclarer) {
  31.         }
  32.     }

4.Topology提交功能

  1. TopologyBuilder topologyBuilder = new TopologyBuilder();
  2.         topologyBuilder.setSpout("DataSourceSpout",new DataSourceSpout());
  3.         topologyBuilder.setBolt("SplitBolt",new SplitBolt()).shuffleGrouping("DataSourceSpout");
  4.         topologyBuilder.setBolt("CountBolt",new CountBolt()).shuffleGrouping("SplitBolt");
  6.         LocalCluster localCluster = new LocalCluster();
  7.         localCluster.submitTopology("LocalWordCountStormTopology",new Config(),topologyBuilder.createTopology());

5.注意

在本案例中，运行完一次后要把文件改名，因为nextTuple()方法一直执行。所以要每发送一个文件后都要给这个文件改名，避免再次发送造成统计错误。