storm wordcount实例

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

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的实时性就体现在这里