Storm 集群的搭建及其Java编程进行简单统计计算 - ToB企服应用市场:ToB评测及商务社交产业平台

version: '2'
services:
zookeeper1:
image: registry.aliyuncs.com/denverdino/zookeeper:3.4.8
container_name: zk1.cloud
environment:
- SERVER_ID=1
- ADDITIONAL_ZOOKEEPER_1=server.1=0.0.0.0:2888:3888
- ADDITIONAL_ZOOKEEPER_2=server.2=zk2.cloud:2888:3888
- ADDITIONAL_ZOOKEEPER_3=server.3=zk3.cloud:2888:3888
zookeeper2:
image: registry.aliyuncs.com/denverdino/zookeeper:3.4.8
container_name: zk2.cloud
environment:
- SERVER_ID=2
- ADDITIONAL_ZOOKEEPER_1=server.1=zk1.cloud:2888:3888
- ADDITIONAL_ZOOKEEPER_2=server.2=0.0.0.0:2888:3888
- ADDITIONAL_ZOOKEEPER_3=server.3=zk3.cloud:2888:3888
zookeeper3:
image: registry.aliyuncs.com/denverdino/zookeeper:3.4.8
container_name: zk3.cloud
environment:
- SERVER_ID=3
- ADDITIONAL_ZOOKEEPER_1=server.1=zk1.cloud:2888:3888
- ADDITIONAL_ZOOKEEPER_2=server.2=zk2.cloud:2888:3888
- ADDITIONAL_ZOOKEEPER_3=server.3=0.0.0.0:2888:3888
ui:
image: registry.aliyuncs.com/denverdino/baqend-storm:1.0.0
command: ui -c nimbus.host=nimbus
environment:
- STORM_ZOOKEEPER_SERVERS=zk1.cloud,zk2.cloud,zk3.cloud
restart: always
container_name: ui
ports:
- 8080:8080
depends_on:
- nimbus
nimbus:
image: registry.aliyuncs.com/denverdino/baqend-storm:1.0.0
command: nimbus -c nimbus.host=nimbus
restart: always
environment:
- STORM_ZOOKEEPER_SERVERS=zk1.cloud,zk2.cloud,zk3.cloud
container_name: nimbus
ports:
- 6627:6627
supervisor:
image: registry.aliyuncs.com/denverdino/baqend-storm:1.0.0
command: supervisor -c nimbus.host=nimbus -c supervisor.slots.ports=[6700,6701,6702,6703]
restart: always
environment:
- affinity:role!=supervisor
- STORM_ZOOKEEPER_SERVERS=zk1.cloud,zk2.cloud,zk3.cloud
depends_on:
- nimbus
networks:
default:
external:
name: zk-net

复制代码

docker pull zookeeper:3.4.8 docker pull storm:1.0.0

复制代码

docker-compose -f storm.yml up -d

复制代码

package bolt;
import java.io.IOException;
import java.util.HashMap;
import java.util.Map;
import java.util.Set;
import org.apache.storm.task.OutputCollector;
import org.apache.storm.task.TopologyContext;
import org.apache.storm.topology.OutputFieldsDeclarer;
import org.apache.storm.topology.base.BaseRichBolt;
import org.apache.storm.tuple.Tuple;
import org.apache.storm.tuple.Values;
@SuppressWarnings("serial")
public class TypeCountBolt extends BaseRichBolt {
OutputCollector collector;
Map<String,Integer> map = new HashMap<String, Integer>();
Map<String,Float> map2 = new HashMap<String, Float>();
public void prepare(Map stormConf, TopologyContext context, OutputCollector collector) {
this.collector = collector;
}
public void execute(Tuple input) {
String line = input.getStringByField("line");
String[] data = line.split(",");
Integer count = map.get(data[2]);
Float total_amount = map2.get(data[2]);
if(count==null){
count = 0;
}
if(total_amount==null){
total_amount = 0.0f;
}
count++;
total_amount+=Float.parseFloat(data[3]) * Integer.parseInt(data[4]);
map.put(data[2],count);
map2.put(data[2],total_amount);
System.out.println("~~~~~~~~~~~~~~~~~~~~~~~");
Set<Map.Entry<String,Integer>> entrySet = map.entrySet();
for(Map.Entry<String,Integer> entry :entrySet){
System.out.println("交易量:");
System.out.println(entry);
}
System.out.println();
Set<Map.Entry<String,Float>> entrySet2 = map2.entrySet();
for(Map.Entry<String,Float> entry :entrySet2){
System.out.println("交易总金额:");
System.out.println(entry);
}
}
public void declareOutputFields(OutputFieldsDeclarer declarer) {
}
}

复制代码

package bolt;
import org.apache.storm.task.OutputCollector;
import org.apache.storm.task.TopologyContext;
import org.apache.storm.topology.OutputFieldsDeclarer;
import org.apache.storm.topology.base.BaseRichBolt;
import org.apache.storm.tuple.Tuple;
import java.text.ParseException;
import java.text.SimpleDateFormat;
import java.util.Date;
import java.util.HashMap;
import java.util.Map;
import java.util.Set;
public class TimeCountBolt extends BaseRichBolt {
OutputCollector collector;
Map<Integer,Integer> map = new HashMap<Integer, Integer>();
Map<Integer,Float> map2 = new HashMap<Integer, Float>();
public void prepare(Map stormConf, TopologyContext context, OutputCollector collector) {
this.collector = collector;
}
public void execute(Tuple input) {
String line = input.getStringByField("line");
String[] data = line.split(",");
Date date = new Date();
SimpleDateFormat dateFormat= new SimpleDateFormat("yyyy-MM-dd hh:mm:ss");
try {
date = dateFormat.parse(data[0]);
} catch (ParseException e) {
e.printStackTrace();
}
Integer count = map.get(date.getHours());
Float total_amount = map2.get(date.getHours());
if(count==null){
count = 0;
}
if(total_amount==null){
total_amount = 0.0f;
}
count++;
total_amount+=Float.parseFloat(data[3]) * Integer.parseInt(data[4]);
map.put(date.getHours(),count);
map2.put(date.getHours(),total_amount);
System.out.println("~~~~~~~~~~~~~~~~~~~~~~~");
Set<Map.Entry<Integer,Integer>> entrySet = map.entrySet();
for(Map.Entry<Integer,Integer> entry :entrySet){
System.out.println("交易量:");
System.out.println(entry);
}
System.out.println();
Set<Map.Entry<Integer,Float>> entrySet2 = map2.entrySet();
for(Map.Entry<Integer,Float> entry :entrySet2){
System.out.println("交易总金额:");
System.out.println(entry);
}
}
public void declareOutputFields(OutputFieldsDeclarer declarer) {
}
}

复制代码