流式计算引擎 —— Storm

作者: 小胡_鸭 | 来源:发表于2022-07-21 14:53 被阅读0次

一、Apache Storm 简介

1、基本概念

Storm 为分布式实时计算提供了一组通用原语，可被用于 “流处理” 之中，实时处理消息并更新数据库。这是管理队列和工作者集群的另一种方式。Storm 也可被用于 “连续计算”（continuous computation），对数据流做连续查询，在计算时就将结果以流的形式输出给用户。它还可被用于 "分布式RPC"，以并行的方式运行昂贵的计算。

Storm 可以方便地在一个计算机集群中编写与扩展复杂的实时运算，Storm 用于实时处理，就好比 Hadoop 用于批处理。Storm 保证每个消息都会得到处理，而且它很快 —— 在一个小集群中，每秒可以处理数以百万计的消息，更棒的是你可以使用任意编程语言来做开发。

2、离线计算和实时计算

实时计算和离线计算，从字面上我们也能看出，这两者主要是在数据处理延迟性上有不同的要求。对应这两种计算模式，有流处理和批处理两种概念。

简单来说，流处理，就是对源源不断的数据流进行处理；而批处理，则是对一定规模量的数据进行计算。流处理要求更高的实时性，而批处理则主要在数据处理规模上发力，批处理的典型代表就是Hadoop MapReduce。

而流处理，经历了几个阶段的发展，Spark Streaming、Storm、Flink等框架，都在流处理方面有不错的变现。

（1）离线计算

离线计算：批量获取数据、批量传输数据、周期性批量计算数据、数据展示。
代表技术：Sqoop 批量导入数据、HDFS 批量存储数据、MapReduce 批量计算、Hive 提供便捷的查询功能。

（2）实时计算

流式计算：数据实时产生、数据实时传输、数据实时计算、实时展示。
代表技术：Flume 实时获取数据、Kafka/metaq 实时数据存储、Storm/JStorm 实时数据计算、Redis 实时结果缓存、mysql 持久化存储。
总结：将源源不断产生的数据实时收集并实时运算，尽可能快的得到计算结果。

总结如下：

计算模式	代表技术	数据采集	数据存储	本质	特点
离线计算	MapReduce、Spark Core、Flink DataSet	Sqoop、Flume	HDFS、HBase	批处理	1、数据量大且时间周期长 2、在大量数据上进行复杂的批量运算 3、数据在计算之前已经固定，不再会发生变化 4、能够方便的查询批量计算的结果
实时计算	Storm、Spark Streaming、Flink DataStream	Flume	Kafka、Redis	源源不断	1、数据实时到达 2、数据到达次序独立，不受应用系统所控制 3、数据规模大且无法预知容量 4、原始数据一经处理，除非特意保存，否则不能被再次取出处理，或者再次提取数据代价昂贵

从目前的发展趋势来看，实时计算正在成为主流。举一个大家都熟悉的例子，你在淘宝上下单了某个商品或者点击浏览了某件商品，你就会发现你的页面立马就会给你推荐这种商品的广告和类似商品的店铺，这背后就是实时计算的功劳。

以蓄水池为例，就是一个典型的实时计算系统，每时每刻都有水进来，再经过沉淀、过滤、吸附、消毒等处理步骤，再供应到用户管道出去。

二、Storm 体系架构

Storm 是一个依赖于 zk 的主从架构，主节点是 nimbus，从节点是 supervisor。

首先 nimbus 负责接收客户端提交的任务（Topology）请求，将其程序包和配置信息保存在本地，并上传至 zk，将任务分配的信息也记录在 zk 中。

每个 supervisor 从节点可以启动多个 worker，通过不同端口来区分同一节点上的不同 worker，每个 worker 上会有一个 executor，execuotor 是 worker 进程中的具体的物理线程，同一个 Spout/Bolt 可能会共享一个物理线程，但是一个 executor 只能运行隶属于同一个 Spout/Bolt 的 task。

三、环境搭建

1、安装配置

# 解压安装
tar -zxvf apache-storm-1.0.3.tar.gz -C ~/training/

# 配置环境变量
vi ~/.bash_profile
STORM_HOME=/root/training/apache-storm-1.0.3
export STORM_HOME

PATH=$STORM_HOME/bin:$PATH
export PATH

source ~/.bash_profile

2、伪分布模式

配置（$STORM_HOME/conf/storm.yaml）

# zk
storm.zookeeper.servers:
      - "bigdata111"

# 主节点
nimbus.seeds: ["bigdata111"]

# 上传jar的时候，上传的路径
storm.local.dir: "/root/training/apache-storm-1.0.3/tmp"

# 表示在一个从节点上启动4个Worker
supervisor.slots:ports:
      - 6700
      - 6701
      - 6702
      - 6703
      
# 每个任务分配一个日志收集器，监控处理的数据结果。默认0      
"topology.eventlogger.executors": 1

先启动 zk，再启动 Storm

# 启动zk
zkServer.sh start

# 启动Storm nimbus
storm nimbus &

# 启动Storm supervisor
storm supervisor &

# 启动Storm UI (web端口)
storm ui &

# 启动Storm日志收集器
storm logviewer &

3、全分布模式

首先保证集群中的系统时间一致

先启动 zk 集群，我部署的zk集群机器在 bigdata112、bigdata113、bigdata114 三台机器上，所以分别在这三台机器上启动 zk

zkServer.sh start

全分布模式的 Storm 集群部署在 bigdata112、bigdata113、big114 组成的集群上，其中 bigdata112、bigdata113 是主节点，用来支持集群的高可用。

在 bigdata112 上安装 storm，并修改配置文件（$STORM_HOME/conf/storm.yaml）

storm.zookeeper.servers:
      - "bigdata112"
      - "bigdata113"
      - "bigdata114"    
nimbus.seeds: ["bigdata112", "bigdata113"]
storm.local.dir: "/root/training/apache-storm-1.0.3/tmp"
supervisor.slots:ports:
      - 6700
      - 6701
      - 6702
      - 6703
"topology.eventlogger.executors": 1

创建目录

mkdir /root/training/apache-storm-1.0.3/tmp

最后从 bigdata112 将安装目录拷贝到 bigdata113、bigdata114

for i in {bigdata113,bigdata114}; do scp -r /root/training/apache-storm-1.0.3 $i:/root/training; done

修改三台机器上的配置文件 ~/.bash_profile，添加环境变量，最后 source ~/.bash_profile

STORM_HOME=/root/training/apache-storm-1.0.3
export STORM_HOME

PATH=$STORM_HOME/bin:$PATH
export PATH

启动集群

# 在 bigdata112 上启动 nimbus、ui、logviewer
storm nimbus &
storm ui &
storm logviewer &

# 在 bigdata113 上启动 nimbus、supervisor
storm nimbus &
storm supervisor &

# 在 bigdata114 上启动 supervisor
storm supervisor &

查看 web 控制台（http://bigdata112:8080/）

可以看到有两个 nimbus，启动 bigdata112 是 Leader，bigdata113 是备份，而 bigdata113、bigdata114 上各有一个 supervisor 进程，其 id 可以在 zk 中看到

将 bigdata112 上的 nimbus 进程杀死，可以看到 bigdata113 成为了新的 Leader。

四、Storm 小程序

1、演示

Storm 自带一些样例小程序，位于目录：$STORM_HOME/examples/storm-starter，有个 README.markdown 对此进行了说明

在该目录下通过执行以下命令，可以提交执行一个自带的 WordCount 小程序

# 提交任务命令格式：
# storm jar [jar路径] [包名.类名] [任务名称]
storm jar storm-starter-topologies-1.0.3.jar org.apache.storm.starter.WordCountTopology MyWC

提交任务时会先将 jar 包上传至 storm 服务器，然后由 nimbus 在 zk 上创建任务，再由 zk 分给 supervisor 上的 worker 执行。

提交成功后，打开控制台可以看到对应的任务

可以打开 debug 开关，这样就可以通过日志实时查看数据处理的情况

可以看到日志里显示了数据的采集，分词和计数的情况。

2、开发

用 storm 开发一个 wordcount 小程序，先来分析一下数据处理的流程，首先在 Storm 中采集数据的组件是 Spout，数据计算处理的组件是 Bolt。

Spout 首先要获取到数据源的数据，并将数据传递给处理链后面的 Bolt，Spout 和 Bolt 之间传递数据时 Tuple，每个 Tuple 就是一个数据行，tuple 是具有 schema 的，所以 tuple = schema + value，在一个组件将数据传递给下一个组件时，必须声明 tuple 的 schema。

由于大数据都是集群多节点多线程并发处理，每个环节处理的组件都有多个实例，所以一个组件采集或处理完数据要传递给下一个组件时，面临要传递给哪个实例的问题，分组策略常见的有：

随机分组
按字段分组 —— 类似于 MapReduce 中数据传递给 Reduce 的过程
广播分组

对于 wordcount 程序来说，首先需要定义一个数据采集组件负责 WordCountSpout，接着对整行数据进行分词，所以需要定义一个 WordCountSplitBolt 组件，最后对单词进行计数需要有一个 WordCountTotalBolt 组件，整个处理的链路就是 WordCountSpout -> WordCountSplitBolt -> WordCountTotalBolt。

如果有需求，比如将处理结果记录到 HDFS、Redis 或 HBase 中，则可以在链路末端定义相应的组件来处理。

（1）Spout

package demo.wordcount;

import org.apache.storm.spout.SpoutOutputCollector;
import org.apache.storm.task.TopologyContext;
import org.apache.storm.topology.OutputFieldsDeclarer;
import org.apache.storm.topology.base.BaseRichSpout;
import org.apache.storm.tuple.Fields;
import org.apache.storm.tuple.Values;
import org.apache.storm.utils.Utils;

import java.util.Map;
import java.util.Random;

/*
 * 模拟产生数据，并且送入下一个bolt组件进行单词的拆分
 */
public class WordCountSpout extends BaseRichSpout {

    private SpoutOutputCollector collector;

    //数据
    private String[] datas = {"I love Beijing",
                              "I love China",
                              "Beijing is the capital of China"};

    @Override
    public void open(Map map, TopologyContext topologyContext, SpoutOutputCollector spoutOutputCollector) {
        // 初始化
        // SpoutOutputCollector代表Spout组件的输出流
        this.collector = spoutOutputCollector;
    }

    @Override
    public void nextTuple() {
        // 如何产生数据？如何输出？
        Utils.sleep(3000); //每三秒产生一条数据

        // 产生一个3以内的随机数
        int index = (new Random()).nextInt(3);
        System.out.println("采集的数据是：" + datas[index]);

        // 发送给下一个组件
        this.collector.emit(new Values(datas[index]));
    }

    // 申明输出的Tuple的格式（schema）
    @Override
    public void declareOutputFields(OutputFieldsDeclarer outputFieldsDeclarer) {
        outputFieldsDeclarer.declare(new Fields("sentence"));
    }
}

（2）Bolt

package demo.wordcount;

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.Fields;
import org.apache.storm.tuple.Tuple;
import org.apache.storm.tuple.Values;

import java.util.Map;

/*
 * 接收上一级组件发来的数据，进行单词的拆分
 */
public class WordCountSplitBolt extends BaseRichBolt {

    //  输出流
    private OutputCollector collector;

    @Override
    public void prepare(Map stormConf, TopologyContext context, OutputCollector collector) {
        // 初始化
        this.collector = collector;
    }

    @Override
    public void execute(Tuple input) {
        // 如何处理上一级组件发送来的数据
        // 得到数据： I love Beijing
        String data = input.getStringByField("sentence");
        String[] words = data.split(" ");

        //输出
        for (String s : words) {
            this.collector.emit(new Values(s, 1));
        }
    }

    @Override
    public void declareOutputFields(OutputFieldsDeclarer declarer) {
        //申明该组件输出的Schema格式
        declarer.declare(new Fields("word","count"));
    }
}

package demo.wordcount;

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.Fields;
import org.apache.storm.tuple.Tuple;
import org.apache.storm.tuple.Values;

import java.util.HashMap;
import java.util.Map;

/*
 * 从上一级组件中接收数据，进行单词的统计
 */
public class WordCountTotalBolt extends BaseRichBolt {

    //  输出流
    private OutputCollector collector;

    // 定义一个集合保存最后的结果
    // 相当于是Redis
    private Map<String, Integer> result = new HashMap<String, Integer>();

    @Override
    public void prepare(Map stormConf, TopologyContext context, OutputCollector collector) {
        this.collector = collector;
    }

    @Override
    public void execute(Tuple tuple) {
        // 如何处理上级组件发来的数据
        // 得到上级组件发来的数据
        String word = tuple.getStringByField("word");
        int count = tuple.getIntegerByField("count");

        // 判断是否是第一次出现这个单词
        if (result.containsKey(word)) {
            // 已经存在
            int total = result.get(word);
            result.put(word, total+count);
        } else {
            // 是第一次出现
            result.put(word, count);
        }

        System.out.println("统计的结果是：" + result);

        // 输出                          单词       计数
        this.collector.emit(new Values(word,result.get(word)));
    }

    @Override
    public void declareOutputFields(OutputFieldsDeclarer declarer) {
        // 输出的格式：Schema
        declarer.declare(new Fields("word", "total"));
    }
}

package demo.wordcount;

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.Fields;
import org.apache.storm.tuple.Tuple;
import org.apache.storm.tuple.Values;

import java.util.HashMap;
import java.util.Map;

/*
 * 从上一级组件中接收数据，进行单词的统计
 */
public class WordCountTotalBolt extends BaseRichBolt {

    //  输出流
    private OutputCollector collector;

    // 定义一个集合保存最后的结果
    // 相当于是Redis
    private Map<String, Integer> result = new HashMap<String, Integer>();

    @Override
    public void prepare(Map stormConf, TopologyContext context, OutputCollector collector) {
        this.collector = collector;
    }

    @Override
    public void execute(Tuple tuple) {
        // 如何处理上级组件发来的数据
        // 得到上级组件发来的数据
        String word = tuple.getStringByField("word");
        int count = tuple.getIntegerByField("count");

        // 判断是否是第一次出现这个单词
        if (result.containsKey(word)) {
            // 已经存在
            int total = result.get(word);
            result.put(word, total+count);
        } else {
            // 是第一次出现
            result.put(word, count);
        }

        System.out.println("统计的结果是：" + result);

        // 输出                          单词       计数
        this.collector.emit(new Values(word,result.get(word)));
    }

    @Override
    public void declareOutputFields(OutputFieldsDeclarer declarer) {
        // 输出的格式：Schema
        declarer.declare(new Fields("word", "total"));
    }
}

（3）Tolopogy

package demo.wordcount;

import org.apache.storm.Config;
import org.apache.storm.LocalCluster;
import org.apache.storm.StormSubmitter;
import org.apache.storm.generated.StormTopology;
import org.apache.storm.topology.TopologyBuilder;
import org.apache.storm.tuple.Fields;

public class WordCountTopology {
    public static void main(String[] args) throws Exception {
        // 创建一个任务
        TopologyBuilder builder = new TopologyBuilder();

        // 设置任务的第一个组件：spout组件
        builder.setSpout("myspout", new WordCountSpout());

        // 设置任务的第二个组件：splitbolt
        builder.setBolt("mysplit", new WordCountSplitBolt()).shuffleGrouping("myspout");//随机分组

        // 设置任务的第三个组件：totalbolt                                       上一级组件       按哪个字段
        builder.setBolt("mytotal", new WordCountTotalBolt()).fieldsGrouping("mysplit", new Fields("word"));

        // 创建一个任务
        StormTopology job = builder.createTopology();
        // 配置参数
        Config conf = new Config();

        // 提交任务：1、本地模式    2、集群模式
        // 方式一：本地模式
//        LocalCluster cluster = new LocalCluster();
//        cluster.submitTopology("mywc", conf, job);

        //方式二：集群模式
        StormSubmitter.submitTopology(args[0], conf, job);
    }
}

五、Storm 与其他组件集成

Storm 还很方便跟其他大数据组件集成，比如有可能你想要将实时计算的结果写入 Redis、HDFS 或 HBase 中。

1、Redis

引入相关的依赖：

<dependency>
    <groupId>redis.clients</groupId>
    <artifactId>jedis</artifactId>
    <version>2.7.0</version>
</dependency>

<!-- https://mvnrepository.com/artifact/org.apache.storm/storm-redis -->
<dependency>
    <groupId>org.apache.storm</groupId>
    <artifactId>storm-redis</artifactId>
    <version>${storm.version}</version>
</dependency>

定义 RedisBolt，方法返回的是一个 **RedisStoreBolt** 匿名对象，getKeyFromTuple() 方法定义了要将 Tuple 里的哪个字段作为 key，getValueFromTuple() 方法指定了要将哪个字段作为 value，最后 RedisDataTypeDescription() 定义了缓存的类型为 Hash，缓存名为 wordcount。

private static IRichBolt createRedisBolt() {
    // 创建RedisStoreBolt
    //创建Redis连接池
    JedisPoolConfig.Builder builder = new JedisPoolConfig.Builder();
    builder.setHost("192.168.190.111");
    builder.setPort(6379);
    JedisPoolConfig config = builder.build();

    // storeMapper表示数据类型
    return new RedisStoreBolt(config, new RedisStoreMapper() {

        @Override
        public RedisDataTypeDescription getDataTypeDescription() {
            // Redis的数据类型
            return new RedisDataTypeDescription(RedisDataTypeDescription.RedisDataType.HASH,"wordcount");
        }

        @Override
        public String getValueFromTuple(ITuple tuple) {
            // 把Tuple中的哪个字段作为Value
            return String.valueOf(tuple.getIntegerByField("total"));
        }

        @Override
        public String getKeyFromTuple(ITuple tuple) {
            // 把Tuple中的哪个字段作为Key
            return tuple.getStringByField("word");
        }
    });
}

在 WordCountTotalBolt 组件后处理数据

// ...

// 设置任务的第三个组件：totalbolt                                上一级组件       按哪个字段
builder.setBolt("mytotal", new WordCountTotalBolt()).fieldsGrouping("mysplit", new Fields("word"));

// 设置任务的第四个组件：redisbolt
builder.setBolt("myredis", createRedisBolt()).shuffleGrouping("mytotal");

// ...

程序准备好后，可以本地执行，也可以打包提交到 Storm 服务器上执行。

启动 Redis 服务器

redis-server

# 查看redis进程
ps -ef | grep redis

启动程序，进入 redis-cli 可以看到已经写入缓存

2、HDFS

引入相关的依赖：

<hadoop.version>2.7.3</hadoop.version>

<!-- 连带引入 $HADOOP_HOME/share/hadoop/hdfs/lib -->
<dependency>
    <groupId>org.apache.hadoop</groupId>
    <artifactId>hadoop-hdfs</artifactId>
    <version>${hadoop.version}</version>
</dependency>

<dependency>
    <groupId>org.apache.hadoop</groupId>
    <artifactId>hadoop-hdfs-nfs</artifactId>
    <version>${hadoop.version}</version>
</dependency>

<!-- 连带引入 $HADOOP_HOME/share/hadoop/hdfs/common -->
<dependency>
    <groupId>org.apache.hadoop</groupId>
    <artifactId>hadoop-common</artifactId>
    <version>${hadoop.version}</version>
</dependency>

<dependency>
    <groupId>org.apache.hadoop</groupId>
    <artifactId>hadoop-nfs</artifactId>
    <version>${hadoop.version}</version>
</dependency>

<!-- https://mvnrepository.com/artifact/org.apache.storm/storm-hdfs -->
<dependency>
    <groupId>org.apache.storm</groupId>
    <artifactId>storm-hdfs</artifactId>
    <version>${storm.version}</version>
</dependency>

定义 HdfsBolt，指定 HDFS namenode地址，数据存储的路径，tuple 字段分隔符等。

private static IRichBolt createHDFSBolt() {
    // 创建一个HDFS的bolt
    HdfsBolt bolt = new HdfsBolt();

    //指定NameNode的地址
    bolt.withFsUrl("hdfs://192.168.190.111:9000");

    //HDFS存储的路径
    bolt.withFileNameFormat(new DefaultFileNameFormat().withPath("/stormdata"));

    //指定key和value的分隔符：Beijing|10
    bolt.withRecordFormat(new DelimitedRecordFormat().withFieldDelimiter("|"));

    //生成文件的策略
    bolt.withRotationPolicy(new FileSizeRotationPolicy(5.0f, FileSizeRotationPolicy.Units.MB));//每5M生成一个文件

    //与HDFS同步策略
    bolt.withSyncPolicy(new CountSyncPolicy(1024));

    return bolt;
}

启动 HDFS，启动程序，可以看到 HDFS 保存了数据文件

3、HBase

引入相关依赖：

<hbase.version>1.3.1</hbase.version>

<dependency>
    <groupId>org.apache.hbase</groupId>
    <artifactId>hbase-client</artifactId>
    <version>${hbase.version}</version>
</dependency>

<dependency>
    <groupId>org.apache.hbase</groupId>
    <artifactId>hbase-server</artifactId>
    <version>${hbase.version}</version>
</dependency>

首先启动 HBase 服务器，然后创建一个表，列族为 info

create table 'myresult','info'

使用原生的 Bolt 子类来实现写入 HBase 的功能（也可以使用 storm-hbase）

package demo.wordcount;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.client.HTable;
import org.apache.hadoop.hbase.client.Put;
import org.apache.hadoop.hbase.util.Bytes;
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.io.IOException;
import java.util.Map;

public class WordCountHBaseBolt extends BaseRichBolt {
    //定义表的客户端
    private HTable table;

    @Override
    public void prepare(Map stormConf, TopologyContext context, OutputCollector collector) {
        //配置ZooKeeper的地址
        Configuration conf = new Configuration();
        conf.set("hbase.zookeeper.quorum", "192.168.190.111");

        //得到表的客户端
        try {
            table = new HTable(conf,"myresult");
        } catch (IOException e) {
            // TODO Auto-generated catch block
            e.printStackTrace();
        }
    }

    @Override
    public void execute(Tuple tuple) {
        //把单词计数的结果存入HBase
        String word = tuple.getStringByField("word");
        int total = tuple.getIntegerByField("total");

        //构造一个Put对象
        Put put = new Put(Bytes.toBytes(word));
        put.add(Bytes.toBytes("info"), Bytes.toBytes("word"), Bytes.toBytes(String.valueOf(total)));

        try {
            table.put(put);
        } catch (IOException e) {
            // TODO Auto-generated catch block
            e.printStackTrace();
        }
    }

    @Override
    public void declareOutputFields(OutputFieldsDeclarer declarer) {

    }
}

使用 word 作为行键，使用 info 列族的 word 字段来存储单词出现次数，启动程序，可以从 HBase shell 中查到保存的数据结果