Storm 集群的搭建及其Java编程进行简单统计计算

一、Storm集群构建
编写storm 与 zookeeper的yml文件

 

storm yml文件的编写

具体如下：

1. version: '2'
3. services:
5.   zookeeper1:
7.     image: registry.aliyuncs.com/denverdino/zookeeper:3.4.8
9.     container_name: zk1.cloud
11.     environment:
13.       - SERVER_ID=1
15.       - ADDITIONAL_ZOOKEEPER_1=server.1=0.0.0.0:2888:3888
17.       - ADDITIONAL_ZOOKEEPER_2=server.2=zk2.cloud:2888:3888
19.       - ADDITIONAL_ZOOKEEPER_3=server.3=zk3.cloud:2888:3888
21.   zookeeper2:
23.     image: registry.aliyuncs.com/denverdino/zookeeper:3.4.8
25.     container_name: zk2.cloud
27.     environment:
29.       - SERVER_ID=2
31.       - ADDITIONAL_ZOOKEEPER_1=server.1=zk1.cloud:2888:3888
33.       - ADDITIONAL_ZOOKEEPER_2=server.2=0.0.0.0:2888:3888
35.       - ADDITIONAL_ZOOKEEPER_3=server.3=zk3.cloud:2888:3888
37.   zookeeper3:
39.     image: registry.aliyuncs.com/denverdino/zookeeper:3.4.8
41.     container_name: zk3.cloud
43.     environment:
45.       - SERVER_ID=3
47.       - ADDITIONAL_ZOOKEEPER_1=server.1=zk1.cloud:2888:3888
49.       - ADDITIONAL_ZOOKEEPER_2=server.2=zk2.cloud:2888:3888
51.       - ADDITIONAL_ZOOKEEPER_3=server.3=0.0.0.0:2888:3888
53.   ui:
55.     image: registry.aliyuncs.com/denverdino/baqend-storm:1.0.0
57.     command: ui -c nimbus.host=nimbus
59.     environment:
61.       - STORM_ZOOKEEPER_SERVERS=zk1.cloud,zk2.cloud,zk3.cloud
63.     restart: always
65.     container_name: ui
67.     ports:
69.       - 8080:8080
71.     depends_on:
73.       - nimbus
75.   nimbus:
77.     image: registry.aliyuncs.com/denverdino/baqend-storm:1.0.0
79.     command: nimbus -c nimbus.host=nimbus
81.     restart: always
83.     environment:
85.       - STORM_ZOOKEEPER_SERVERS=zk1.cloud,zk2.cloud,zk3.cloud
87.     container_name: nimbus
89.     ports:
91.       - 6627:6627
93.   supervisor:
95.     image: registry.aliyuncs.com/denverdino/baqend-storm:1.0.0
97.     command: supervisor -c nimbus.host=nimbus -c supervisor.slots.ports=[6700,6701,6702,6703]
99.     restart: always
101.     environment:
103.       - affinity:role!=supervisor
105.       - STORM_ZOOKEEPER_SERVERS=zk1.cloud,zk2.cloud,zk3.cloud
107.     depends_on:
109.       - nimbus
111. networks:
113.   default:
115.     external:
117.       name: zk-net

复制代码

 
 
拉取Storm搭建需要的镜像，这里我选择镜像版本为 zookeeper:3.4.8  storm:1.0.0
键入命令：

1. docker pull zookeeper:3.4.8  docker pull storm:1.0.0

复制代码

 

storm镜像 获取

使用docker-compose 构建集群
在power shell中执行以下命令：
 

1. docker-compose -f storm.yml up -d

复制代码

 

                                                        　　　　　　　　　　　　　　　　　　　　　　docker-compose 构建集群
在浏览器中打开localhost:8080 可以看到storm集群的详细情况

 

storm UI 展示

二、Storm统计任务
统计股票交易情况交易量和交易总金额   （数据文件存储在csv文件中）
编写DataSourceSpout类

 

DataSourceSpout类

编写bolt类
 
 
 
编写topology类
 
 
需要注意的是 Storm Java API 下有本地模型和远端模式
在本地模式下的调试不依赖于集群环境，可以进行简单的调试
如果需要使用生产模式，则需要将
1、 编写和自身业务相关的spout和bolt类，并将其打包成一个jar包
 
2、将上述的jar包放到客户端代码能读到的任何位置，
 
3、使用如下方式定义一个拓扑（Topology）
 
 
演示结果：
本地模式下的调试：
 
正在执行：

 

根据24小时

 

 

根据股票种类

 
 
生产模式：

 

向集群提交topology

                                                       
 
 
 
三、核心计算bolt的代码
1.统计不同类型的股票交易量和交易总金额：

1. package bolt;
5. import java.io.IOException;
7. import java.util.HashMap;
9. import java.util.Map;
11. import java.util.Set;
15. import org.apache.storm.task.OutputCollector;
17. import org.apache.storm.task.TopologyContext;
19. import org.apache.storm.topology.OutputFieldsDeclarer;
21. import org.apache.storm.topology.base.BaseRichBolt;
23. import org.apache.storm.tuple.Tuple;
25. import org.apache.storm.tuple.Values;
29. @SuppressWarnings("serial")
31. public class TypeCountBolt extends BaseRichBolt {
35.     OutputCollector collector;
39.     Map<String,Integer> map = new HashMap<String, Integer>();
43.     Map<String,Float> map2 = new HashMap<String, Float>();
49.     public void prepare(Map stormConf, TopologyContext context, OutputCollector collector) {
51.         this.collector = collector;
55.     }
59.     public void execute(Tuple input) {
61.         String line = input.getStringByField("line");
63.         String[] data = line.split(",");
65.         Integer count = map.get(data[2]);
67.         Float total_amount = map2.get(data[2]);
69.         if(count==null){
71.             count = 0;
73.         }
75.         if(total_amount==null){
77.             total_amount = 0.0f;
79.         }
81.         count++;
83.         total_amount+=Float.parseFloat(data[3]) * Integer.parseInt(data[4]);
85.         map.put(data[2],count);
87.         map2.put(data[2],total_amount);
91.         System.out.println("~~~~~~~~~~~~~~~~~~~~~~~");
93.         Set<Map.Entry<String,Integer>> entrySet = map.entrySet();
95.         for(Map.Entry<String,Integer> entry :entrySet){
97.             System.out.println("交易量:");
99.             System.out.println(entry);
101.         }
103.         System.out.println();
105.         Set<Map.Entry<String,Float>> entrySet2 = map2.entrySet();
107.         for(Map.Entry<String,Float> entry :entrySet2){
109.             System.out.println("交易总金额:");
111.             System.out.println(entry);
113.         }
115.     }
121.     public void declareOutputFields(OutputFieldsDeclarer declarer) {
123.     }
127. }

复制代码

 
 
2. 统计不同每个小时的交易量和交易总金额

1. package bolt;
5. import org.apache.storm.task.OutputCollector;
7. import org.apache.storm.task.TopologyContext;
9. import org.apache.storm.topology.OutputFieldsDeclarer;
11. import org.apache.storm.topology.base.BaseRichBolt;
13. import org.apache.storm.tuple.Tuple;
17. import java.text.ParseException;
19. import java.text.SimpleDateFormat;
21. import java.util.Date;
23. import java.util.HashMap;
25. import java.util.Map;
27. import java.util.Set;
31. public  class TimeCountBolt extends BaseRichBolt {
33.     OutputCollector collector;
37.     Map<Integer,Integer> map = new HashMap<Integer, Integer>();
41.     Map<Integer,Float> map2 = new HashMap<Integer, Float>();
47.     public void prepare(Map stormConf, TopologyContext context, OutputCollector collector) {
49.         this.collector = collector;
53.     }
57.     public void execute(Tuple input) {
59.         String line = input.getStringByField("line");
61.         String[] data = line.split(",");
65.         Date date = new Date();
67.         SimpleDateFormat dateFormat= new SimpleDateFormat("yyyy-MM-dd hh:mm:ss");
69.         try {
71.             date = dateFormat.parse(data[0]);
73.         } catch (ParseException e) {
75.             e.printStackTrace();
77.         }
81.         Integer count = map.get(date.getHours());
83.         Float total_amount = map2.get(date.getHours());
85.         if(count==null){
87.             count = 0;
89.         }
91.         if(total_amount==null){
93.             total_amount = 0.0f;
95.         }
97.         count++;
99.         total_amount+=Float.parseFloat(data[3]) * Integer.parseInt(data[4]);
101.         map.put(date.getHours(),count);
103.         map2.put(date.getHours(),total_amount);
107.         System.out.println("~~~~~~~~~~~~~~~~~~~~~~~");
109.         Set<Map.Entry<Integer,Integer>> entrySet = map.entrySet();
111.         for(Map.Entry<Integer,Integer> entry :entrySet){
113.             System.out.println("交易量:");
115.             System.out.println(entry);
117.         }
119.         System.out.println();
121.         Set<Map.Entry<Integer,Float>> entrySet2 = map2.entrySet();
123.         for(Map.Entry<Integer,Float> entry :entrySet2){
125.             System.out.println("交易总金额:");
127.             System.out.println(entry);
129.         }
131.     }
137.     public void declareOutputFields(OutputFieldsDeclarer declarer) {
139.     }
141. }

复制代码

 

免责声明：如果侵犯了您的权益，请联系站长，我们会及时删除侵权内容，谢谢合作！