Storm本地开发案例—总和计算与单词统计

￡神魔★判官ぃ 2021-11-22 23:12 192阅读 0赞

## 概述 ##

在Strom的API中提供了`LocalCluster`对象，这样在不用搭建Storm环境或者Storm集群的情况下也能够开发Storm的程序，非常方便。

基于Maven构建工程项目，其所需要的依赖如下：

<dependency>
        <groupId>org.apache.storm</groupId>
        <artifactId>storm-core</artifactId>
        <version>1.0.2</version>
    </dependency>

## Storm本地开发案例1：总和计算 ##

### 需求分析 ###

需求如下：

数据源不断产生递增数字，对产生的数字累加求和

分析如下：

Strom的Topology包含Spout和Bolt两种节点类型，在这个案例中，可以使用Spout来对数据源进行处理（模拟产生数据），
    然后将其发送到计算和的Bolt中，所以实际上这里只需要使用一个Spout节点和一个Bolt节点就可以了。

### 程序开发 ###

在理解了Storm的设计思想后，将其与MapReduce的设计思想进行对比，再看下面的程序代码其实是非常好理解的。

OrderSpout

/**
         * 数据源
         */
    static class OrderSpout extends BaseRichSpout {
    
        private Map conf;   // 当前组件配置信息
        private TopologyContext context;    // 当前组件上下文对象
        private SpoutOutputCollector collector; // 发送tuple的组件
    
        @Override
        public void open(Map conf, TopologyContext context, SpoutOutputCollector collector) {
            this.conf = conf;
            this.context = context;
            this.collector = collector;
        }
    
        /**
             * 接收数据的核心方法
             */
        @Override
        public void nextTuple() {
            long num = 0;
            while (true) {
                num++;
                StormUtil.sleep(1000);
                System.out.println("当前时间" + StormUtil.df_yyyyMMddHHmmss.format(new Date()) + "产生的订单金额：" + num);
                this.collector.emit(new Values(num));
            }
        }
    
        /**
             * 是对发送出去的数据的描述schema
             */
        @Override
        public void declareOutputFields(OutputFieldsDeclarer declarer) {
            declarer.declare(new Fields("order_cost"));
        }
    }

SumBolt

private Long sumOrderCost = 0L;
    
    /**
         * 计算和的Bolt节点
         */
    static class SumBolt extends BaseRichBolt {
    
        private Map conf;   // 当前组件配置信息
        private TopologyContext context;    // 当前组件上下文对象
        private OutputCollector collector; // 发送tuple的组件
    
        @Override
        public void prepare(Map stormConf, TopologyContext context, OutputCollector collector) {
            this.conf = conf;
            this.context = context;
            this.collector = collector;
        }
    
        private Long sumOrderCost = 0L;
    
        /**
             * 处理数据的核心方法
             */
        @Override
        public void execute(Tuple input) {
            Long orderCost = input.getLongByField("order_cost");
            sumOrderCost += orderCost;
    
            System.out.println("商城网站到目前" + StormUtil.df_yyyyMMddHHmmss.format(new Date()) + "的商品总交易额" + sumOrderCost);
            StormUtil.sleep(1000);
        }
    
        /**
             * 如果当前bolt为最后一个处理单元，该方法可以不用管
             */
        @Override
        public void declareOutputFields(OutputFieldsDeclarer declarer) {
    
        }
    }

StormLocalSumTopology

/**
     * 1°、实现数字累加求和的案例：数据源不断产生递增数字，对产生的数字累加求和。
     * <p>
     * Storm组件：Spout、Bolt、数据是Tuple，使用main中的Topology将spout和bolt进行关联
     * MapReduce的组件：Mapper和Reducer、数据是Writable，通过一个main中的job将二者关联
     * <p>
     * 适配器模式（Adapter）：BaseRichSpout，其对继承接口中一些没必要的方法进行了重写，但其重写的代码没有实现任何功能。
     *                        我们称这为适配器模式
     */
    public class StormLocalSumTopology {
        /**
         * 构建拓扑，相当于在MapReduce中构建Job
         */
        public static void main(String[] args) {
            TopologyBuilder builder = new TopologyBuilder();
            /**
             * 设置spout和bolt的dag（有向无环图）
             */
            builder.setSpout("id_order_spout", new OrderSpout());
            builder.setBolt("id_sum_bolt", new SumBolt())
                    .shuffleGrouping("id_order_spout"); // 通过不同的数据流转方式，来指定数据的上游组件
            // 使用builder构建topology
            StormTopology topology = builder.createTopology();
            // 启动topology
            LocalCluster localCluster = new LocalCluster(); // 本地开发模式，创建的对象为LocalCluster
            String topologyName = StormLocalSumTopology.class.getSimpleName();  // 拓扑的名称
            Config config = new Config();   // Config()对象继承自HashMap，但本身封装了一些基本的配置
            localCluster.submitTopology(topologyName, config, topology);
        }
    }

> 需要说明的是，Spout和Bolt的类都作为StormLocalSumTopology的静态成员变量，这样做是为了开发的方便，当然实际上也可以将其单独作为一个文件。

### 测试 ###

执行主函数，其输出如下：

当前时间20180412213836产生的订单金额：1
    商城网站到目前20180412213836的商品总交易额1
    当前时间20180412213837产生的订单金额：2
    商城网站到目前20180412213837的商品总交易额3
    当前时间20180412213838产生的订单金额：3
    商城网站到目前20180412213838的商品总交易额6
    ......

## Storm本地开发案例2：单词统计 ##

### 需求分析 ###

需求如下：

监控一个目录下的文件，当发现有新文件的时候，把文件读取过来，解析文件中的内容，统计单词出现的总次数

分析如下：

可以设置三个节点：
    Spout：用于持续读取目录下需要被监听（通过后缀名标识）的文件，并且将每一行输出到下一个Bolt中
            （类似于MapReduce中的FileInputFormat）
    Bolt1：读取行，并解析其中的单词，将每个单词输出到下一个Bolt中
            （类似于MapReduce中的Mapper）
    Bolt2：读取单词，进行统计计算
            （类似于MapReduce中的Reducer）

### 程序开发 ###

FileSpout

/**
         * Spout，获取数据源，这里是持续读取某一目录下的文件，并将每一行输出到下一个Bolt中
         */
    static class FileSpout extends BaseRichSpout {
        private Map conf;   // 当前组件配置信息
        private TopologyContext context;    // 当前组件上下文对象
        private SpoutOutputCollector collector; // 发送tuple的组件
    
        @Override
        public void open(Map conf, TopologyContext context, SpoutOutputCollector collector) {
            this.conf = conf;
            this.context = context;
            this.collector = collector;
        }
    
        @Override
        public void nextTuple() {
            File directory = new File("D:/data/storm");
            // 第二个参数extensions的意思就是，只采集某些后缀名的文件
            Collection<File> files = FileUtils.listFiles(directory, new String[]{"txt"}, true);
            for (File file : files) {
                try {
                    List<String> lines = FileUtils.readLines(file, "utf-8");
                    for(String line : lines) {
                        this.collector.emit(new Values(line));
                    }
                    // 当前文件被消费之后，需要重命名，同时为了防止相同文件的加入，重命名后的文件加了一个随机的UUID，或者加入时间戳也可以的
                    File destFile = new File(file.getAbsolutePath() + "_" + UUID.randomUUID().toString() + ".completed");
                    FileUtils.moveFile(file, destFile);
                } catch (IOException e) {
                    e.printStackTrace();
                }
            }
        }
    
        @Override
        public void declareOutputFields(OutputFieldsDeclarer declarer) {
            declarer.declare(new Fields("line"));
        }
    }

SplitBolt

/**
         * Bolt节点，将接收到的每一行数据切割为一个个单词并发送到下一个节点
         */
    static class SplitBolt extends BaseRichBolt {
    
        private Map conf;   // 当前组件配置信息
        private TopologyContext context;    // 当前组件上下文对象
        private OutputCollector collector; // 发送tuple的组件
    
        @Override
        public void prepare(Map stormConf, TopologyContext context, OutputCollector collector) {
            this.conf = conf;
            this.context = context;
            this.collector = collector;
        }
    
        @Override
        public void execute(Tuple input) {
            String line = input.getStringByField("line");
            String[] words = line.split(" ");
            for (String word : words) {
                this.collector.emit(new Values(word,1));
            }
        }
    
        @Override
        public void declareOutputFields(OutputFieldsDeclarer declarer) {
            declarer.declare(new Fields("word", "count"));
        }
    }

WCBolt

/**
         * Bolt节点，执行单词统计计算
         */
    static class WCBolt extends BaseRichBolt {
    
        private Map conf;   // 当前组件配置信息
        private TopologyContext context;    // 当前组件上下文对象
        private OutputCollector collector; // 发送tuple的组件
    
        @Override
        public void prepare(Map stormConf, TopologyContext context, OutputCollector collector) {
            this.conf = conf;
            this.context = context;
            this.collector = collector;
        }
    
        private Map<String, Integer> map = new HashMap<>();
    
        @Override
        public void execute(Tuple input) {
            String word = input.getStringByField("word");
            Integer count = input.getIntegerByField("count");
            /*if (map.containsKey(word)) {
                    map.put(word, map.get(word) + 1);
                } else {
                    map.put(word, 1);
                }*/
            map.put(word, map.getOrDefault(word, 0) + 1);
    
            System.out.println("====================================");
            map.forEach((k ,v)->{
                System.out.println(k + ":::" +v);
            });
        }
    
        @Override
        public void declareOutputFields(OutputFieldsDeclarer declarer) {
    
        }
    }

StormLocalWordCountTopology

/**
     * 2°、单词计数：监控一个目录下的文件，当发现有新文件的时候，
            把文件读取过来，解析文件中的内容，统计单词出现的总次数
            E:\data\storm
     */
    public class StormLocalWordCountTopology {
    
        /**
         * 构建拓扑，组装Spout和Bolt节点，相当于在MapReduce中构建Job
         */
        public static void main(String[] args) {
            TopologyBuilder builder = new TopologyBuilder();
            // dag
            builder.setSpout("id_file_spout", new FileSpout());
            builder.setBolt("id_split_bolt", new SplitBolt()).shuffleGrouping("id_file_spout");
            builder.setBolt("id_wc_bolt", new WCBolt()).shuffleGrouping("id_split_bolt");
    
            StormTopology stormTopology = builder.createTopology();
            LocalCluster cluster = new LocalCluster();
            String topologyName = StormLocalWordCountTopology.class.getSimpleName();
            Config config = new Config();
            cluster.submitTopology(topologyName, config, stormTopology);
        }
    }

### 测试 ###

执行程序后，往目标目录中添加`.txt`文件，程序输出如下：

====================================
    hello:::1
    ====================================
    hello:::1
    you:::1
    ====================================
    hello:::2
    you:::1
    ====================================
    hello:::2
    he:::1
    you:::1
    ====================================
    hello:::3
    he:::1
    you:::1
    ====================================
    me:::1
    hello:::3
    he:::1
    you:::1

## Storm名词术语解释 ##

在编写了Storm的程序后，再来看看其相关的术语就容易理解很多了。

*  Topology

Topology用于封装一个实时计算应用程序的逻辑，类似于Hadoop的MapReduce Job

*  Stream消息流

Stream 消息流，是一个没有边界的tuple序列，这些tuples会被以一种分布式的方式并行地创建和处理

*  Spouts消息源

Spouts 消息源，是消息生产者，他会从一个外部源读取数据并向topology里面面发出消息：tuple

*  Bolts消息处理者

Bolts 消息处理者，所有的消息处理逻辑被封装在bolts里面，处理输入的数据流并产生新的输出数据流,
    可执行过滤，聚合，查询数据库等操作

*  Task

Task 每一个Spout和Bolt会被当作很多task在整个集群里面执行,每一个task对应到一个线程.