Storm笔记整理（二）：Storm本地开发案例—总和计算与单词统计

发布时间：2020-08-18 16:18:06 作者：xpleaf
来源：网络阅读：7802

[TOC]

概述

在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对应到一个线程.

Stream groupings 消息分发策略

Stream groupings 消息分发策略,定义一个Topology的其中一步是定义每个tuple接受什么样的流作为输入,
stream grouping就是用来定义一个stream应该如何分配给Bolts们.

Storm笔记整理（二）：Storm本地开发案例—总和计算与单词统计

概述

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

需求分析

程序开发

OrderSpout

SumBolt

StormLocalSumTopology

测试

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

需求分析

程序开发

FileSpout

SplitBolt

WCBolt

StormLocalWordCountTopology

测试

Storm名词术语解释

相关阅读