Storm详解二、写第一个Storm应用

在全面介绍Storm之前，我们先通过一个简单的Demo让大家整体感受一下什么是Storm。

Storm运行模式：

1. 本地模式(Local Mode)： 即Topology（相当于一个任务，后续会详细讲解）  运行在本地机器的单一JVM上，这个模式主要用来开发、调试。
2. 远程模式(Remote Mode):在这个模式，我们把我们的Topology提交到集群，在这个模式中，Storm的所有组件都是线程安全的，因为它们都会运行在不同的Jvm或物理机器上，这个模式就是正式的生产模式。

写一个HelloWord Storm

     我们现在创建这么一个应用，统计文本文件中的单词个数，详细学习过Hadoop的朋友都应该写过。那么我们需要具体创建这样一个Topology，用一个spout负责读取文本文件，用第一个bolt来解析成单词，用第二个bolt来对解析出的单词计数，整体结构如图所示：

可以从这里下载源码：https://github.com/storm-book/examples-ch02-getting_started/zipball/master

     写一个可运行的Demo很简单，我们只需要三步：

1. 创建一个Spout读取数据
2. 创建bolt处理数据
3. 创建一个Topology提交到集群

下面我们就写一下，以下代码拷贝到eclipse（依赖的jar包到官网下载即可）即可运行。

1.创建一个Spout作为数据源

Spout作为数据源，它实现了IRichSpout接口，功能是读取一个文本文件并把它的每一行内容发送给bolt。

[java] view plaincopy

1. package storm.demo.spout;
2. import java.io.BufferedReader;
3. import java.io.FileNotFoundException;
4. import java.io.FileReader;
5. import java.util.Map;
6. import backtype.storm.spout.SpoutOutputCollector;
7. import backtype.storm.task.TopologyContext;
8. import backtype.storm.topology.IRichSpout;
9. import backtype.storm.topology.OutputFieldsDeclarer;
10. import backtype.storm.tuple.Fields;
11. import backtype.storm.tuple.Values;
12. public class WordReader implements IRichSpout {
13. private static final long serialVersionUID = 1L;
14. private SpoutOutputCollector collector;
15. private FileReader fileReader;
16. private boolean completed = false;
17. public boolean isDistributed() {
18. return false;
19. }
20. /**
21. * 这是第一个方法，里面接收了三个参数，第一个是创建Topology时的配置，
22. * 第二个是所有的Topology数据，第三个是用来把Spout的数据发射给bolt
23. * **/
24. @Override
25. public void open(Map conf, TopologyContext context,
26. SpoutOutputCollector collector) {
27. try {
28. //获取创建Topology时指定的要读取的文件路径
29. this.fileReader = new FileReader(conf.get("wordsFile").toString());
30. } catch (FileNotFoundException e) {
31. throw new RuntimeException("Error reading file ["
32. + conf.get("wordFile") + "]");
33. }
34. //初始化发射器
35. this.collector = collector;
36. }
37. /**
38. * 这是Spout最主要的方法，在这里我们读取文本文件，并把它的每一行发射出去（给bolt）
39. * 这个方法会不断被调用，为了降低它对CPU的消耗，当任务完成时让它sleep一下
40. * **/
41. @Override
42. public void nextTuple() {
43. if (completed) {
44. try {
45. Thread.sleep(1000);
46. } catch (InterruptedException e) {
47. // Do nothing
48. }
49. return;
50. }
51. String str;
52. // Open the reader
53. BufferedReader reader = new BufferedReader(fileReader);
54. try {
55. // Read all lines
56. while ((str = reader.readLine()) != null) {
57. /**
58. * 发射每一行，Values是一个ArrayList的实现
59. */
60. this.collector.emit(new Values(str), str);
61. }
62. } catch (Exception e) {
63. throw new RuntimeException("Error reading tuple", e);
64. } finally {
65. completed = true;
66. }
67. }
68. @Override
69. public void declareOutputFields(OutputFieldsDeclarer declarer) {
70. declarer.declare(new Fields("line"));
71. }
72. @Override
73. public void close() {
74. // TODO Auto-generated method stub
75. }
76. @Override
77. public void activate() {
78. // TODO Auto-generated method stub
79. }
80. @Override
81. public void deactivate() {
82. // TODO Auto-generated method stub
83. }
84. @Override
85. public void ack(Object msgId) {
86. System.out.println("OK:" + msgId);
87. }
88. @Override
89. public void fail(Object msgId) {
90. System.out.println("FAIL:" + msgId);
91. }
92. @Override
93. public Map<String, Object> getComponentConfiguration() {
94. // TODO Auto-generated method stub
95. return null;
96. }
97. }

2.创建两个bolt来处理Spout发射出的数据

Spout已经成功读取文件并把每一行作为一个tuple(在Storm数据以tuple的形式传递)发射过来，我们这里需要创建两个bolt分别来负责解析每一行和对单词计数。

Bolt中最重要的是execute方法，每当一个tuple传过来时它便会被调用。

第一个bolt：WordNormalizer

[java] view plaincopy

1. package storm.demo.bolt;
2. import java.util.ArrayList;
3. import java.util.List;
4. import java.util.Map;
5. import backtype.storm.task.OutputCollector;
6. import backtype.storm.task.TopologyContext;
7. import backtype.storm.topology.IRichBolt;
8. import backtype.storm.topology.OutputFieldsDeclarer;
9. import backtype.storm.tuple.Fields;
10. import backtype.storm.tuple.Tuple;
11. import backtype.storm.tuple.Values;
12. public class WordNormalizer implements IRichBolt {
13. private OutputCollector collector;
14. @Override
15. public void prepare(Map stormConf, TopologyContext context,
16. OutputCollector collector) {
17. this.collector = collector;
18. }
19. /**这是bolt中最重要的方法，每当接收到一个tuple时，此方法便被调用
20. * 这个方法的作用就是把文本文件中的每一行切分成一个个单词，并把这些单词发射出去（给下一个bolt处理）
21. * **/
22. @Override
23. public void execute(Tuple input) {
24. String sentence = input.getString(0);
25. String[] words = sentence.split(" ");
26. for (String word : words) {
27. word = word.trim();
28. if (!word.isEmpty()) {
29. word = word.toLowerCase();
30. // Emit the word
31. List a = new ArrayList();
32. a.add(input);
33. collector.emit(a, new Values(word));
34. }
35. }
36. //确认成功处理一个tuple
37. collector.ack(input);
38. }
39. @Override
40. public void declareOutputFields(OutputFieldsDeclarer declarer) {
41. declarer.declare(new Fields("word"));
42. }
43. @Override
44. public void cleanup() {
45. // TODO Auto-generated method stub
46. }
47. @Override
48. public Map<String, Object> getComponentConfiguration() {
49. // TODO Auto-generated method stub
50. return null;
51. }
52. }

第二个bolt:WordCounter

[java] view plaincopy

1. package storm.demo.bolt;
2. import java.util.HashMap;
3. import java.util.Map;
4. import backtype.storm.task.OutputCollector;
5. import backtype.storm.task.TopologyContext;
6. import backtype.storm.topology.IRichBolt;
7. import backtype.storm.topology.OutputFieldsDeclarer;
8. import backtype.storm.tuple.Tuple;
9. public class WordCounter implements IRichBolt {
10. Integer id;
11. String name;
12. Map<String, Integer> counters;
13. private OutputCollector collector;
14. @Override
15. public void prepare(Map stormConf, TopologyContext context,
16. OutputCollector collector) {
17. this.counters = new HashMap<String, Integer>();
18. this.collector = collector;
19. this.name = context.getThisComponentId();
20. this.id = context.getThisTaskId();
21. }
22. @Override
23. public void execute(Tuple input) {
24. String str = input.getString(0);
25. if (!counters.containsKey(str)) {
26. counters.put(str, 1);
27. } else {
28. Integer c = counters.get(str) + 1;
29. counters.put(str, c);
30. }
31. // 确认成功处理一个tuple
32. collector.ack(input);
33. }
34. /**
35. * Topology执行完毕的清理工作，比如关闭连接、释放资源等操作都会写在这里
36. * 因为这只是个Demo，我们用它来打印我们的计数器
37. * */
38. @Override
39. public void cleanup() {
40. System.out.println("-- Word Counter [" + name + "-" + id + "] --");
41. for (Map.Entry<String, Integer> entry : counters.entrySet()) {
42. System.out.println(entry.getKey() + ": " + entry.getValue());
43. }
44. counters.clear();
45. }
46. @Override
47. public void declareOutputFields(OutputFieldsDeclarer declarer) {
48. // TODO Auto-generated method stub
49. }
50. @Override
51. public Map<String, Object> getComponentConfiguration() {
52. // TODO Auto-generated method stub
53. return null;
54. }
55. }

3.在main函数中创建一个Topology

在这里我们要创建一个Topology和一个LocalCluster对象，还有一个Config对象做一些配置。

[java] view plaincopy

1. package storm.demo;
2. import storm.demo.bolt.WordCounter;
3. import storm.demo.bolt.WordNormalizer;
4. import storm.demo.spout.WordReader;
5. import backtype.storm.Config;
6. import backtype.storm.LocalCluster;
7. import backtype.storm.topology.TopologyBuilder;
8. import backtype.storm.tuple.Fields;
9. public class WordCountTopologyMain {
10. public static void main(String[] args) throws InterruptedException {
11. //定义一个Topology
12. TopologyBuilder builder = new TopologyBuilder();
13. builder.setSpout("word-reader",new WordReader());
14. builder.setBolt("word-normalizer", new WordNormalizer())
15. .shuffleGrouping("word-reader");
16. builder.setBolt("word-counter", new WordCounter(),2)
17. .fieldsGrouping("word-normalizer", new Fields("word"));
18. //配置
19. Config conf = new Config();
20. conf.put("wordsFile", "d:/text.txt");
21. conf.setDebug(false);
22. //提交Topology
23. conf.put(Config.TOPOLOGY_MAX_SPOUT_PENDING, 1);
24. //创建一个本地模式cluster
25. LocalCluster cluster = new LocalCluster();
26. cluster.submitTopology("Getting-Started-Toplogie", conf,
27. builder.createTopology());
28. Thread.sleep(1000);
29. cluster.shutdown();
30. }
31. }

运行这个函数我们即可看到后台打印出来的单词个数。

（ps:因为是Local模式，运行开始可能会打印很多错误log，这个先不用管）