Storm安装和使用Demo

悠悠 2022-06-05 01:40 143阅读 0赞

http://blog.csdn.net/sqh201030412/article/details/78426812

一,Storm集群的安装  
  
1.下载安装包:

wget [http://archive.apache.org/dist/storm/apache-storm-0.9.7/apache-storm-0.9.7.tar.gz][http_archive.apache.org_dist_storm_apache-storm-0.9.7_apache-storm-0.9.7.tar.gz]

2.解压移动到软件安装目录

tar -zxvf xxxx

3.修改配置文件, conf目录下:

vim storm.yaml

storm.zookeeper.servers:

\- "hadoop6"

\- "hadoop7"

\- "hadoop8"

nimbus.host: "hadoop5"

nimbus.childopts: "-Xmx1024m"

supervisor.childopts: "-Xmx1024m"

worker.childopts: "-Xmx1024m"

ui.childopts: "-Xmx1024m"

supervisor.slots.ports:

\- 6700

\- 6701

\- 6702

\- 6703

4.分发到运行的节点上:

scp -r storm [hadoop@hadoop8][hadoop_hadoop8]:/home/hadoop/app/

......

5.在各个节点启动服务:

*  在nimbus.host所属的机器上启动nimbus服务

cd /export/servers/storm/bin/

nohup ./storm nimbus &

*  在nimbus.host所属的机器上启动ui服务

cd /export/servers/storm/bin/

nohup ./storm ui &

*  在其它个点击上启动supervisor服务

cd /export/servers/storm/bin/

nohup ./storm supervisor &

6.查看

访问nimbus.host:/8080，即可看到storm的ui界面。

二,Storm得使用Demo

1.官方统计字数例子

bin/storm jar examples/storm-starter/storm-starter-topologies-0.9.7.jar storm.starter.WordCountTopology wordcount

2.运行结果

![Center][]

3.自己书写一个字符统计的Demo

\[java\] [view plain][] [copy][view plain]

1.  **package** com.demo.storm.wordcount; 
2.  
3.  **import** java.util.HashMap; 
4.  **import** java.util.Map; 
5.  
6.  **import** org.apache.storm.Config; 
7.  **import** org.apache.storm.StormSubmitter; 
8.  **import** org.apache.storm.generated.AlreadyAliveException; 
9.  **import** org.apache.storm.generated.AuthorizationException; 
10. **import** org.apache.storm.generated.InvalidTopologyException; 
11. **import** org.apache.storm.spout.SpoutOutputCollector; 
12. **import** org.apache.storm.task.OutputCollector; 
13. **import** org.apache.storm.task.TopologyContext; 
14. **import** org.apache.storm.topology.OutputFieldsDeclarer; 
15. **import** org.apache.storm.topology.TopologyBuilder; 
16. **import** org.apache.storm.topology.base.BaseRichBolt; 
17. **import** org.apache.storm.topology.base.BaseRichSpout; 
18. **import** org.apache.storm.tuple.Fields; 
19. **import** org.apache.storm.tuple.Tuple; 
20. **import** org.apache.storm.tuple.Values; 
21. 
22. /\*\*
23.  \* @Description: Storm 单词统计主类
24.  \* @author: songqinghu
25.  \* @date: 2017年11月2日 下午2:18:19
26.  \* Version:1.0
27.  \*/
28. **public****class** WordCountStart \{ 
29. 
30. 
31. **public****static****void** main(String\[\] args) **throws** AlreadyAliveException, InvalidTopologyException, AuthorizationException \{ 
32. //建造者
33.  TopologyBuilder builder = **new** TopologyBuilder(); 
34. //构建spout模块
35.  builder.setSpout("mySpout", **new** MyWordSpout(), Integer.valueOf(4)); 
36. //构建bolt模块 --切割 定义 mySpout为随机发送规则
37.  builder.setBolt("mySplitBolt", **new** MySplitWordBolt(), Integer.valueOf(5)).shuffleGrouping("mySpout"); 
38. //构建bolt模块 --统计 定义 mySplitBolt为Hash分发
39.  builder.setBolt("myCountBolt", **new** MyCountWordBolt(), Integer.valueOf(5)).fieldsGrouping("mySplitBolt", **new** Fields("word")); 
40. 
41.  Config config = **new** Config(); 
42. 
43.  config.setNumWorkers(Integer.valueOf(3)); 
44. //向集群提交任务
45. **if**(args.length==1)\{ 
46.  StormSubmitter.submitTopology(args\[0\],config, builder.createTopology()); 
47.  \}**else**\{ 
48.  StormSubmitter.submitTopology("mywordcount",config, builder.createTopology()); 
49.  \} 
50.  \} 
51. \} 
52. /\*\*
53.  \* @Description: 对接受到的数据进行处理 --统计
54.  \* @author: songqinghu
55.  \* @date: 2017年11月2日 下午2:42:33
56.  \* Version:1.0
57.  \*/
58. **class** MyCountWordBolt **extends** BaseRichBolt \{ 
59. 
60. 
61. **private** Map<String, Integer> wordCount = **new** HashMap<String,Integer>(); 
62. 
63. @Override
64. **public****void** prepare(Map stormConf, TopologyContext context, OutputCollector collector) \{ 
65. 
66.  \} 
67. 
68. @Override
69. **public****void** execute(Tuple input) \{ 
70.  String word = input.getString(0); 
71. 
72. **if**(wordCount.containsKey(word))\{ 
73.  wordCount.put(word, wordCount.get(word)+1); 
74.  \}**else**\{ 
75.  wordCount.put(word, 1); 
76.  \} 
77. 
78.  System.out.println(wordCount); 
79.  \} 
80. 
81. @Override
82. **public****void** declareOutputFields(OutputFieldsDeclarer declarer) \{ 
83. //不在向下发送
84.  \} 
85. 
86. \} 
87. 
88. 
89. /\*\*
90.  \* @Description: 对接受到的数据进行处理 --切割 分组发送
91.  \* @author: songqinghu
92.  \* @date: 2017年11月2日 下午2:42:33
93.  \* Version:1.0
94.  \*/
95. **class** MySplitWordBolt **extends** BaseRichBolt \{ 
96. 
97. **private** OutputCollector collector; 
98. 
99. @Override
100. **public****void** prepare(Map stormConf, TopologyContext context, OutputCollector collector) \{ 
101. **this**.collector =collector; 
102.  \} 
103. 
104. @Override
105. **public****void** execute(Tuple input) \{ 
106.  String words = input.getString(0); 
107.  String\[\] split = words.split(" "); 
108. **for** (String word : split) \{ 
109.  collector.emit(**new** Values(word.trim())); 
110.  \} 
111.  \} 
112. 
113. @Override
114. **public****void** declareOutputFields(OutputFieldsDeclarer declarer) \{ 
115.  declarer.declare(**new** Fields("word")); 
116.  \} 
117. 
118. \} 
119. 
120. 
121. 
122. /\*\*
123.  \* @Description: 数据源接入类,完成数据源的接入和传递工作
124.  \* @author: songqinghu
125.  \* @date: 2017年11月2日 下午2:41:23
126.  \* Version:1.0
127.  \*/
128. **class** MyWordSpout **extends** BaseRichSpout\{ 
129. 
130. **private** SpoutOutputCollector collector; 
131. 
132. 
133. @Override
134. **public****void** open(Map conf, TopologyContext context, SpoutOutputCollector collector) \{ 
135. //初始执行 
136. **this**.collector = collector; 
137.  \} 
138. 
139. @Override
140. **public****void** nextTuple() \{ 
141. //发送数据
142.  collector.emit(**new** Values("My life is in these books Read these and know my heart")); 
143.  \} 
144. 
145. 
146. @Override
147. **public****void** declareOutputFields(OutputFieldsDeclarer declarer) \{ 
148. //定义发送的tuple对应字段
149.  declarer.declare(**new** Fields("words")); 
150.  \} 
151. 
152. \}

4.打包运行即可:

bin/storm jar wordcount.jar com.demo.storm.wordcount.WordCountStart mywordcount

[http_archive.apache.org_dist_storm_apache-storm-0.9.7_apache-storm-0.9.7.tar.gz]: http://archive.apache.org/dist/storm/apache-storm-0.9.7/apache-storm-0.9.7.tar.gz
[hadoop_hadoop8]: mailto:hadoop@hadoop8
[Center]: /images/20220605/41cf69d7f3c14cccb07dc347a651a0ce.png
[view plain]: http://blog.csdn.net/sqh201030412/article/details/78426812#