当前位置: 代码迷 >> 综合 >> Storm Worldcount
  详细解决方案

Storm Worldcount

热度:97   发布时间:2023-12-25 23:17:20.0
在storm环境部署完毕,并正确启动之后,现在就可以真正进入storm开发了,按照惯例,以wordcount作为开始。
这个例子很简单,核心组件包括:一个spout,两个bolt,一个Topology。
spout从一个路径读取文件,然后readLine,向bolt发射,一个文件处理完毕后,重命名,以不再重复处理。
第一个bolt将从spout接收到的字符串按空格split,产生word,发射给下一个bolt。
第二个bolt接收到word后,统计、计数,放到HashMap容器中。

1,定义一个spout,作用是源源不断滴向bolt发射字符串。

点击(此处)折叠或打开

  1. import java.io.File;
  2. import java.io.IOException;
  3. import java.util.Collection;
  4. import java.util.List;
  5. import java.util.Map;

  6. import org.apache.commons.io.FileUtils;
  7. import org.apache.commons.io.filefilter.FileFilterUtils;

  8. import backtype.storm.spout.SpoutOutputCollector;
  9. import backtype.storm.task.TopologyContext;
  10. import backtype.storm.topology.OutputFieldsDeclarer;
  11. import backtype.storm.topology.base.BaseRichSpout;
  12. import backtype.storm.tuple.Fields;
  13. import backtype.storm.tuple.Values;

  14. public class WordReader extends BaseRichSpout {
  15.     private static final long serialVersionUID = 2197521792014017918L;
  16.     private String inputPath;
  17.     private SpoutOutputCollector collector;

  18.     @Override
  19.     @SuppressWarnings(\"rawtypes\")
  20.     public void open(Map conf, TopologyContext context, SpoutOutputCollector collector) {
  21.         this.collector = collector;
  22.         inputPath = (String) conf.get(\"INPUT_PATH\");
  23.     }

  24.     @Override
  25.     public void nextTuple() {
  26.         Collection<File> files = FileUtils.listFiles(new File(inputPath),
  27.                 FileFilterUtils.notFileFilter(FileFilterUtils.suffixFileFilter(\".bak\")), null);
  28.         for (File f : files) {
  29.             try {
  30.                 List<String> lines = FileUtils.readLines(f, \"UTF-8\");
  31.                 for (String line : lines) {
  32.                     collector.emit(new Values(line));
  33.                 }
  34.                 FileUtils.moveFile(f, new File(f.getPath() + System.currentTimeMillis() + \".bak\"));
  35.             } catch (IOException e) {
  36.                 e.printStackTrace();
  37.             }
  38.         }
  39.     }

  40.     @Override
  41.     public void declareOutputFields(OutputFieldsDeclarer declarer) {
  42.         declarer.declare(new Fields(\"line\"));
  43.     }

  44. }

2,定义一个bolt,作用是接收spout发过来的字符串,并分割成word,发射给下一个bolt。

点击(此处)折叠或打开

  1. import org.apache.commons.lang.StringUtils;

  2. import backtype.storm.topology.BasicOutputCollector;
  3. import backtype.storm.topology.OutputFieldsDeclarer;
  4. import backtype.storm.topology.base.BaseBasicBolt;
  5. import backtype.storm.tuple.Fields;
  6. import backtype.storm.tuple.Tuple;
  7. import backtype.storm.tuple.Values;

  8. public class WordSpliter extends BaseBasicBolt {

  9.     private static final long serialVersionUID = -5653803832498574866L;

  10.     @Override
  11.     public void execute(Tuple input, BasicOutputCollector collector) {
  12.         String line = input.getString(0);
  13.         String[] words = line.split(\" \");
  14.         for (String word : words) {
  15.             word = word.trim();
  16.             if (StringUtils.isNotBlank(word)) {
  17.                 word = word.toLowerCase();
  18.                 collector.emit(new Values(word));
  19.             }
  20.         }
  21.     }

  22.     @Override
  23.     public void declareOutputFields(OutputFieldsDeclarer declarer) {
  24.         declarer.declare(new Fields(\"word\"));

  25.     }

  26. }

3,定义一个bolt,接收word,并统计。

点击(此处)折叠或打开

  1. import java.util.HashMap;
  2. import java.util.Map;
  3. import java.util.Map.Entry;

  4. import backtype.storm.task.TopologyContext;
  5. import backtype.storm.topology.BasicOutputCollector;
  6. import backtype.storm.topology.OutputFieldsDeclarer;
  7. import backtype.storm.topology.base.BaseBasicBolt;
  8. import backtype.storm.tuple.Tuple;

  9. public class WordCounter extends BaseBasicBolt {
  10.     private static final long serialVersionUID = 5683648523524179434L;
  11.     private HashMap<String, Integer> counters = new HashMap<String, Integer>();
  12.     private volatile boolean edit = false;

  13.     @Override
  14.     @SuppressWarnings(\"rawtypes\")
  15.     public void prepare(Map stormConf, TopologyContext context) {
  16.         final long timeOffset = Long.parseLong(stormConf.get(\"TIME_OFFSET\").toString());
  17.         new Thread(new Runnable() {
  18.             @Override
  19.             public void run() {
  20.                 while (true) {
  21.                     if (edit) {
  22.                         for (Entry<String, Integer> entry : counters.entrySet()) {
  23.                             System.out.println(entry.getKey() + \" : \" + entry.getValue());
  24.                         }
  25.                         System.out.println(\"WordCounter---------------------------------------\");
  26.                         edit = false;
  27.                     }
  28.                     try {
  29.                         Thread.sleep(timeOffset * 1000);
  30.                     } catch (InterruptedException e) {
  31.                         e.printStackTrace();
  32.                     }
  33.                 }
  34.             }
  35.         }).start();
  36.     }

  37.     @Override
  38.     public void execute(Tuple input, BasicOutputCollector collector) {
  39.         String str = input.getString(0);
  40.         if (!counters.containsKey(str)) {
  41.             counters.put(str, 1);
  42.         } else {
  43.             Integer c = counters.get(str) + 1;
  44.             counters.put(str, c);
  45.         }
  46.         edit = true;
  47.         System.out.println(\"WordCounter+++++++++++++++++++++++++++++++++++++++++++\");
  48.     }

  49.     @Override
  50.     public void declareOutputFields(OutputFieldsDeclarer declarer) {

  51.     }
  52. }
注意 WordCounter类 prepare方法,里面定义了一个Thread,持续监控容器的变化(word个数增加或者新增word)。

4,定义一个Topology,提交作业。

点击(此处)折叠或打开

  1. public class WordCountTopo {
  2.     public static void main(String[] args) {
  3.         if (args.length != 2) {
  4.             System.err.println(\"Usage: inputPaht timeOffset\");
  5.             System.err.println(\"such as : java -jar WordCount.jar D://input/ 2\");
  6.             System.exit(2);
  7.         }
  8.         TopologyBuilder builder = new TopologyBuilder();
  9.         builder.setSpout(\"word-reader\", new WordReader());
  10.         builder.setBolt(\"word-spilter\", new WordSpliter()).shuffleGrouping(\"word-reader\");
  11.         builder.setBolt(\"word-counter\", new WordCounter()).shuffleGrouping(\"word-spilter\");
  12.         String inputPaht = args[0];
  13.         String timeOffset = args[1];
  14.         Config conf = new Config();
  15.         conf.put(\"INPUT_PATH\", inputPaht);
  16.         conf.put(\"TIME_OFFSET\", timeOffset);
  17.         conf.setDebug(false);
  18.         LocalCluster cluster = new LocalCluster();
  19.         cluster.submitTopology(\"WordCount\", conf, builder.createTopology());
  20.     }
  21. }

5,代码完成后,导出jar(导出时不要指定Main class),然后上传至storm集群,通过命令./storm jar com.x.x.WordCountTopo /data/tianzhen/input 2来提交作业。
Topo启动,spout、bolt执行过程:


Thread监控的统计结果:


源文件处理之后被重命名为*.bak。

和Hadoop不同,在任务执行完之后,Topo不会停止,spout会一直监控数据源,不停地往bolt发射数据。
所以现在如果源数据发生变化,应该能够立马体现出来。我往path下再放一个文本文件,结果:

可见,结果立刻更新了,storm的实时性就体现在这里。