Storm的WordCount在Storm集群上运行

爱被打了一巴掌 2022-12-27 14:25 236阅读 0赞

# 前提： #

### 运行在storm集群里的log日志不会打印在屏幕上，因此需要写一个OSUtil类进行**nc -lk 8888**端口的socket打印！！！ ###

**OSUtil.java：**

package storm.util;
    
    import java.io.IOException;
    import java.io.OutputStream;
    import java.lang.management.ManagementFactory;
    import java.lang.management.RuntimeMXBean;
    import java.net.InetAddress;
    import java.net.Socket;
    import java.net.UnknownHostException;
    import java.nio.ByteBuffer;
    
    /**
     * @author: wtl
     * @License: (C) Copyright 2020, wtl Corporation Limited.
     * @Contact: 1050100468@qq.com
     * @Date: 2020/12/12 8:33
     * @Version: 1.0
     * @Description:
     */
    public class OSUtil {
        private static String getHostName() throws Exception {
            return InetAddress.getLocalHost().getHostName();
        }
    
        private static String getIpAddress() throws Exception {
            return InetAddress.getLocalHost().getHostAddress();
        }
    
        private static String getPid(){
            RuntimeMXBean runtime = ManagementFactory.getRuntimeMXBean();
            String name = runtime.getName();
            String pid = name.substring(0, name.indexOf("@"));
            return name + ":" + pid;
        }
    
        private static long getCurrentThreadId(){
            return Thread.currentThread().getId();
        }
    
    
        private static String info(Object object,String message) throws Exception {
            return getHostName() + " " + getIpAddress() + " " + getPid() + " " + getCurrentThreadId() + " " + object.hashCode() + "=>" + message;
        }
    
        public static void sendMessage(Object object,String message) {
            try {
                Socket socket = new Socket("hadoop04",8888);
                OutputStream outputStream = socket.getOutputStream();
                outputStream.write((info(object,message) + "\r\n").getBytes());
                outputStream.flush();
                outputStream.close();
                socket.close();
            }
            catch (Exception e){
                e.printStackTrace();
            }
        }
    }

** WordCountSpout.java：**

package storm;
    
    import lombok.SneakyThrows;
    import org.apache.storm.spout.SpoutOutputCollector;
    import org.apache.storm.task.TopologyContext;
    import org.apache.storm.topology.IRichSpout;
    import org.apache.storm.topology.OutputFieldsDeclarer;
    import org.apache.storm.tuple.Fields;
    import org.apache.storm.tuple.Values;
    import storm.util.OSUtil;
    import sun.nio.ch.IOUtil;
    
    import java.util.Map;
    import java.util.Random;
    import java.util.concurrent.TimeUnit;
    
    /**
     * @author: wtl
     * @License: (C) Copyright 2020, wtl Corporation Limited.
     * @Contact: 1050100468@qq.com
     * @Date: 2020/11/27 5:14
     * @Version: 1.0
     * @Description:
     */
    public class WordCountSpout implements IRichSpout {
    
        private SpoutOutputCollector spoutOutputCollector;
        private String [] lines = {"I love you","hello world","hello kitty","nihao beijing"};
        private Random random = new Random();
    
        @Override
        public void open(Map<String, Object> map, TopologyContext topologyContext, SpoutOutputCollector spoutOutputCollector) {
    
            this.spoutOutputCollector = spoutOutputCollector;
        }
    
        @Override
        public void close() {
    
        }
    
        @Override
        public void activate() {
    
        }
    
        @Override
        public void deactivate() {
    
        }
    
        @SneakyThrows
        @Override
        public void nextTuple() {
            OSUtil.sendMessage(this,"nextTuple()");
            int nextInt = random.nextInt(lines.length);
            spoutOutputCollector.emit(new Values(lines[nextInt]));
            TimeUnit.SECONDS.sleep(3);
        }
    
        @Override
        public void ack(Object o) {
    
        }
    
        @Override
        public void fail(Object o) {
    
        }
    
        @Override
        public void declareOutputFields(OutputFieldsDeclarer outputFieldsDeclarer) {
            outputFieldsDeclarer.declare(new Fields("line"));
        }
    
        @Override
        public Map<String, Object> getComponentConfiguration() {
            return null;
        }
    }

**SplitBolt.java： **

package storm;
    
    import org.apache.storm.task.OutputCollector;
    import org.apache.storm.task.TopologyContext;
    import org.apache.storm.topology.IRichBolt;
    import org.apache.storm.topology.OutputFieldsDeclarer;
    import org.apache.storm.tuple.Fields;
    import org.apache.storm.tuple.Tuple;
    import org.apache.storm.tuple.Values;
    import storm.util.OSUtil;
    
    import java.util.Map;
    import java.util.stream.Stream;
    
    /**
     * @author: wtl
     * @License: (C) Copyright 2020, wtl Corporation Limited.
     * @Contact: 1050100468@qq.com
     * @Date: 2020/11/27 5:41
     * @Version: 1.0
     * @Description:
     */
    public class SplitBolt implements IRichBolt {
    
        private TopologyContext topologyContext;
        private OutputCollector outputCollector;
    
        @Override
        public void prepare(Map<String, Object> map, TopologyContext topologyContext, OutputCollector outputCollector) {
            OSUtil.sendMessage(this,"SplitBolt.prepare()");
            this.topologyContext = topologyContext;
            this.outputCollector = outputCollector;
        }
    
        @Override
        public void execute(Tuple tuple) {
            String line = tuple.getStringByField("line");
            String[] splits = line.split(" ");
    
            Stream.of(splits).forEach(word -> {
                outputCollector.emit(new Values(word,1));
            });
        }
    
        @Override
        public void cleanup() {
    
        }
    
        @Override
        public void declareOutputFields(OutputFieldsDeclarer outputFieldsDeclarer) {
            outputFieldsDeclarer.declare(new Fields("word","count"));
        }
    
        @Override
        public Map<String, Object> getComponentConfiguration() {
            return null;
        }
    }

** TotalBolt.java：**

package storm;
    
    import org.apache.storm.task.OutputCollector;
    import org.apache.storm.task.TopologyContext;
    import org.apache.storm.topology.IRichBolt;
    import org.apache.storm.topology.OutputFieldsDeclarer;
    import org.apache.storm.tuple.Fields;
    import org.apache.storm.tuple.Tuple;
    import org.apache.storm.tuple.Values;
    import storm.util.OSUtil;
    
    import java.util.HashMap;
    import java.util.Map;
    
    /**
     * @author: wtl
     * @License: (C) Copyright 2020, wtl Corporation Limited.
     * @Contact: 1050100468@qq.com
     * @Date: 2020/11/27 5:57
     * @Version: 1.0
     * @Description:
     */
    public class TotalBolt implements IRichBolt {
        private TopologyContext topologyContext;
        private OutputCollector outputCollector;
        private static Map<String,Integer> map = new HashMap<>();;
    
        @Override
        public void prepare(Map<String, Object> map, TopologyContext topologyContext, OutputCollector outputCollector) {
            OSUtil.sendMessage(this,"TotalBolt.prepare()");
            this.topologyContext = topologyContext;
            this.outputCollector = outputCollector;
        }
    
        @Override
        public void execute(Tuple tuple) {
            String word = tuple.getStringByField("word");
            Integer count = tuple.getIntegerByField("count");
            if (!map.containsKey(word)){
                map.put(word,count);
            }
            else{
                map.put(word,map.get(word) + count);
            }
    
            map.forEach((key, value) -> {
                //System.out.println(key + ": " + value);
                OSUtil.sendMessage(this,key + ": " + value);
            });
        }
    
        @Override
        public void cleanup() {
        }
    
        @Override
        public void declareOutputFields(OutputFieldsDeclarer outputFieldsDeclarer) {
            outputFieldsDeclarer.declare(new Fields("word","total"));
        }
    
        @Override
        public Map<String, Object> getComponentConfiguration() {
            return null;
        }
    }

### 在hadoop04机器上执行： ###

nc -lk 8888

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3d0bDE5OTI_size_16_color_FFFFFF_t_70][]

## 测试： ##

### 1、接收blot的数据，开了两个线程 ###

topologyBuilder.setSpout("wcSpout",new WordCountSpout(),2).setNumTasks(2);

**SplitBolt.java： **

package storm;
    
    import org.apache.storm.task.OutputCollector;
    import org.apache.storm.task.TopologyContext;
    import org.apache.storm.topology.IRichBolt;
    import org.apache.storm.topology.OutputFieldsDeclarer;
    import org.apache.storm.tuple.Fields;
    import org.apache.storm.tuple.Tuple;
    import org.apache.storm.tuple.Values;
    import storm.util.OSUtil;
    
    import java.util.Map;
    import java.util.stream.Stream;
    
    /**
     * @author: wtl
     * @License: (C) Copyright 2020, wtl Corporation Limited.
     * @Contact: 1050100468@qq.com
     * @Date: 2020/11/27 5:41
     * @Version: 1.0
     * @Description:
     */
    public class SplitBolt implements IRichBolt {
    
        private TopologyContext topologyContext;
        private OutputCollector outputCollector;
    
        @Override
        public void prepare(Map<String, Object> map, TopologyContext topologyContext, OutputCollector outputCollector) {
            OSUtil.sendMessage(this,"SplitBolt.prepare()");
            this.topologyContext = topologyContext;
            this.outputCollector = outputCollector;
        }
    
        @Override
        public void execute(Tuple tuple) {
            String line = tuple.getStringByField("line");
            OSUtil.sendMessage(this,line);
            String[] splits = line.split(" ");
    
            Stream.of(splits).forEach(word -> {
                outputCollector.emit(new Values(word,1));
            });
        }
    
        @Override
        public void cleanup() {
    
        }
    
        @Override
        public void declareOutputFields(OutputFieldsDeclarer outputFieldsDeclarer) {
            outputFieldsDeclarer.declare(new Fields("word","count"));
        }
    
        @Override
        public Map<String, Object> getComponentConfiguration() {
            return null;
        }
    }

**测试结果：**

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3d0bDE5OTI_size_16_color_FFFFFF_t_70 1][]

2、测试

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3d0bDE5OTI_size_16_color_FFFFFF_t_70]: /images/20221120/85e34e64255e481cbe6c609e929f7057.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3d0bDE5OTI_size_16_color_FFFFFF_t_70 1]: /images/20221120/8aaf2ac24b3f4fde99b233f49902b227.png