Storm 入门的Demo教程

- 日理万妓 2022-05-29 04:36 198阅读 0赞

## Storm介绍 ##

Storm是Twitter开源的分布式实时大数据处理框架，最早开源于github，从0.9.1版本之后，归于Apache社区，被业界称为实时版Hadoop。随着越来越多的场景对Hadoop的MapReduce高延迟无法容忍，比如网站统计、推荐系统、预警系统、金融系统(高频交易、股票)等等，大数据实时处理解决方案（流计算）的应用日趋广泛，目前已是分布式技术领域最新爆发点，而Storm更是流计算技术中的佼佼者和主流。

### Storm的核心组件 ###

*  **Nimbus**：即Storm的Master，负责资源分配和任务调度。一个Storm集群只有一个Nimbus。
 *  **Supervisor**：即Storm的Slave，负责接收Nimbus分配的任务，管理所有Worker，一个Supervisor节点中包含多个Worker进程。
 *  **Worker**：工作进程，每个工作进程中都有多个Task。
 *  **Task**：任务，在 Storm 集群中每个 Spout 和 Bolt 都由若干个任务（tasks）来执行。每个任务都与一个执行线程相对应。
 *  **Topology**：计算拓扑，Storm 的拓扑是对实时计算应用逻辑的封装，它的作用与 MapReduce 的任务（Job）很相似，区别在于 MapReduce 的一个 Job 在得到结果之后总会结束，而拓扑会一直在集群中运行，直到你手动去终止它。拓扑还可以理解成由一系列通过数据流（Stream Grouping）相互关联的 Spout 和 Bolt 组成的的拓扑结构。
 *  **Stream**：数据流（Streams）是 Storm 中最核心的抽象概念。一个数据流指的是在分布式环境中并行创建、处理的一组元组（tuple）的无界序列。数据流可以由一种能够表述数据流中元组的域（fields）的模式来定义。
 *  **Spout**：数据源（Spout）是拓扑中数据流的来源。一般 Spout 会从一个外部的数据源读取元组然后将他们发送到拓扑中。根据需求的不同，Spout 既可以定义为可靠的数据源，也可以定义为不可靠的数据源。一个可靠的 Spout能够在它发送的元组处理失败时重新发送该元组，以确保所有的元组都能得到正确的处理；相对应的，不可靠的 Spout 就不会在元组发送之后对元组进行任何其他的处理。一个 Spout可以发送多个数据流。
 *  **Bolt**：拓扑中所有的数据处理均是由 Bolt 完成的。通过数据过滤（filtering）、函数处理（functions）、聚合（aggregations）、联结（joins）、数据库交互等功能，Bolt 几乎能够完成任何一种数据处理需求。一个 Bolt 可以实现简单的数据流转换，而更复杂的数据流变换通常需要使用多个 Bolt 并通过多个步骤完成。
 *  **Stream grouping**：为拓扑中的每个 Bolt 的确定输入数据流是定义一个拓扑的重要环节。数据流分组定义了在 Bolt 的不同任务（tasks）中划分数据流的方式。在 Storm 中有八种内置的数据流分组方式。
 *  **Reliability**：可靠性。Storm 可以通过拓扑来确保每个发送的元组都能得到正确处理。通过跟踪由 Spout 发出的每个元组构成的元组树可以确定元组是否已经完成处理。每个拓扑都有一个“消息延时”参数，如果 Storm 在延时时间内没有检测到元组是否处理完成，就会将该元组标记为处理失败，并会在稍后重新发送该元组。

Storm程序再Storm集群中运行的示例图如下:  
![这里写图片描述][Image 1]

### Topology ###

为什么把Topology单独提出来呢，因为Topology是我们开发程序主要的用的组件。  
Topology和MapReduce很相像。  
MapReduce是Map进行获取数据，Reduce进行处理数据。  
而Topology则是使用Spout获取数据，Bolt来进行计算。  
总的来说就是一个Topology由一个或者多个的Spout和Bolt组成。

具体流程是怎么走，可以通过查看下面这张图来进行了解。  
示例图:  
![这里写图片描述][Image 1]

注:图片来源[http://www.tianshouzhi.com/api/tutorials/storm/52][http_www.tianshouzhi.com_api_tutorials_storm_52]。

图片有三种模式，解释如下:  
第一种比较简单，就是由一个Spout获取数据，然后交给一个Bolt进行处理;  
第二种稍微复杂点，由一个Spout获取数据，然后交给一个Bolt进行处理一部分，然后在交给下一个Bolt进行处理其他部分。  
第三种则比较复杂，一个Spout可以同时发送数据到多个Bolt，而一个Bolt也可以接受多个Spout或多个Bolt，最终形成多个数据流。但是这种数据流必须是有方向的，有起点和终点，不然会造成死循环，数据永远也处理不完。就是Spout发给Bolt1，Bolt1发给Bolt2，Bolt2又发给了Bolt1,最终形成了一个环状。

## Storm 集群安装 ##

之前已经写过了，这里就不在说明了。  
博客地址:[http://www.panchengming.com/2018/01/26/pancm70/][http_www.panchengming.com_2018_01_26_pancm70]

## Storm Hello World ##

前面讲了一些Storm概念，可能在理解上不太清楚，那么这里我们就用一个Hello World代码示例来体验下Storm运作的流程吧。

### 环境准备 ###

在进行代码开发之前，首先得做好相关的准备。  
本项目是使用Maven构建的，使用Storm的版本为1.1.1。  
Maven的相关依赖如下:

<dependency>
        <groupId>org.apache.storm</groupId>
        <artifactId>storm-core</artifactId>
        <version>1.1.1</version>
        <scope>provided</scope>
     </dependency>

### 具体流程 ###

在写代码的时候，我们先来明确要用Storm做什么。  
那么第一个程序，就简单的输出下信息。  
具体步骤如下:  
1. 启动topology，设置好Spout和Bolt。  
2. 将Spout获取的数据传递给Bolt。  
3. Bolt接受Spout的数据进行打印。

#### Spout ####

那么首先开始编写Spout类。一般是实现 **IRichSpout** 或继承**BaseRichSpout**该类，然后实现该方法。  
这里我们继承BaseRichSpout这个类,该类需要实现这几个主要的方法:

##### 一、open #####

open()方法中是在ISpout接口中定义，在Spout组件初始化时被调用。  
有三个参数，它们的作用分别是:  
1. Storm配置的Map;  
2. topology中组件的信息;  
3. 发射tuple的方法;

代码示例:

@Override
        public void open(Map map, TopologyContext arg1, SpoutOutputCollector collector) {
            System.out.println("open:"+map.get("test"));
            this.collector = collector;
        }

##### 二、nextTuple #####

nextTuple()方法是Spout实现的核心。  
也就是主要执行方法，用于输出信息,通过`collector.emit`方法发射。

这里我们的数据信息已经写死了，所以这里我们就直接将数据进行发送。  
这里设置只发送两次。  
代码示例:

@Override
        public void nextTuple() {
            if(count<=2){
                System.out.println("第"+count+"次开始发送数据...");
                this.collector.emit(new Values(message));
            }
            count++;
        }

##### 三、declareOutputFields #####

declareOutputFields是在IComponent接口中定义，用于声明数据格式。  
即输出的一个Tuple中，包含几个字段。

因为这里我们只发射一个，所以就指定一个。如果是多个，则用逗号隔开。  
代码示例:

@Override
        public void declareOutputFields(OutputFieldsDeclarer declarer) {
            System.out.println("定义格式...");
            declarer.declare(new Fields(field));
        }

##### 四、ack #####

ack是在ISpout接口中定义，用于表示Tuple处理成功。

代码示例:

@Override
        public void ack(Object obj) {
            System.out.println("ack:"+obj);
        }

##### 五、fail #####

fail是在ISpout接口中定义，用于表示Tuple处理失败。

代码示例:

@Override
        public void fail(Object obj) {
            System.out.println("失败:"+obj);
        }

##### 六、close #####

close是在ISpout接口中定义，用于表示Topology停止。

代码示例:

@Override
        public void close() {
            System.out.println("关闭...");
        }

至于还有其他的，这里就不在一一列举了。

#### Bolt ####

Bolt是用于处理数据的组件，主要是由execute方法来进行实现。一般来说需要实现 **IRichBolt** 或继承**BaseRichBolt**该类，然后实现其方法。  
需要实现方法如下:

##### 一、prepare #####

在Bolt启动前执行，提供Bolt启动环境配置的入口。  
参数基本和Sqout一样。  
一般对于不可序列化的对象进行实例化。  
这里的我们就简单的打印下

@Override
        public void prepare(Map map, TopologyContext arg1, OutputCollector collector) {
            System.out.println("prepare:"+map.get("test"));
            this.collector=collector;
        }

**注:如果是可以序列化的对象，那么最好是使用构造函数。**

##### 二、execute #####

execute()方法是Bolt实现的核心。  
也就是执行方法，每次Bolt从流接收一个订阅的tuple，都会调用这个方法。  
从tuple中获取消息可以使用 `tuple.getString()`和`tuple.getStringByField();`这两个方法。个人推荐第二种，可以通过field来指定接收的消息。  
**注:如果继承的是IRichBolt，则需要手动ack。这里就不用了,BaseRichBolt会自动帮我们应答。**  
代码示例:

@Override
        public void execute(Tuple tuple) {
    // String msg=tuple.getString(0);
            String msg=tuple.getStringByField("test");
            //这里我们就不做消息的处理，只打印
            System.out.println("Bolt第"+count+"接受的消息:"+msg); 
            count++;
            /** * * 没次调用处理一个输入的tuple，所有的tuple都必须在一定时间内应答。 * 可以是ack或者fail。否则，spout就会重发tuple。 */
    // collector.ack(tuple);
        }

##### 三、declareOutputFields #####

和Spout的一样。  
因为到了这里就不再输出了，所以就什么都没写。

@Override
        public void declareOutputFields(OutputFieldsDeclarer arg0) {        
        }

##### cleanup #####

cleanup是IBolt接口中定义,用于释放bolt占用的资源。  
Storm在终止一个bolt之前会调用这个方法。  
因为这里没有什么资源需要释放，所以就简单的打印一句就行了。

@Override
        public void cleanup() {
            System.out.println("资源释放");
        }

#### Topology ####

这里我们就是用main方法进行提交topology。  
不过在提交topology之前，需要进行相应的设置。  
这里我就不一一细说了，代码的注释已经很详细了。  
代码示例:

import org.apache.storm.Config;
        import org.apache.storm.LocalCluster;
        import org.apache.storm.StormSubmitter;
        import org.apache.storm.topology.TopologyBuilder;
    
        /** * * Title: App * Description: * storm测试 * Version:1.0.0 * @author pancm * @date 2018年3月6日 */
        public class App { 
    
            private static final String str1="test1"; 
            private static final String str2="test2"; 
    
            public static void main(String[] args)  {
                // TODO Auto-generated method stub
                //定义一个拓扑
                TopologyBuilder builder=new TopologyBuilder();
                //设置一个Executeor(线程)，默认一个
                builder.setSpout(str1, new TestSpout());
                //设置一个Executeor(线程)，和一个task
                builder.setBolt(str2, new TestBolt(),1).setNumTasks(1).shuffleGrouping(str1);
                Config conf = new Config();
                conf.put("test", "test");
                try{
                  //运行拓扑
               if(args !=null&&args.length>0){ //有参数时，表示向集群提交作业，并把第一个参数当做topology名称
                 System.out.println("远程模式");
                     StormSubmitter.submitTopology(args[0], conf, builder.createTopology());
              } else{
       //没有参数时，本地提交
            //启动本地模式
             System.out.println("本地模式");
            LocalCluster cluster = new LocalCluster();
            cluster.submitTopology("111" ,conf,  builder.createTopology() );
            Thread.sleep(10000);
        // 关闭本地集群
            cluster.shutdown();
              }
                }catch (Exception e){
                    e.printStackTrace();
                }   
            }
        }

运行该方法，输出结果如下:

本地模式
    定义格式...
    open:test
    第1次开始发送数据...
    第2次开始发送数据...
    prepare:test
    Bolt第1接受的消息:这是个测试消息!
    Bolt第2接受的消息:这是个测试消息!
    资源释放
    关闭...

到这里，是不是基本上对Storm的运作有些了解了呢。  
这个demo达到了上述的三种模式图中的第一种，一个Spout传输数据， 一个Bolt处理数据。

那么如果我们想达到第二种模式呢，那又该如何做呢？  
假如我们想统计下在一段文本中的单词出现频率的话，我们只需执行一下步骤就可以了。  
1.首先将Spout中的message消息进行更改为数组，并依次将消息发送到TestBolt。  
2.然后TestBolt将获取的数据进行分割，将分割的数据发送到TestBolt2。  
3.TestBolt2对数据进行统计，在程序关闭的时候进行打印。  
4.Topology成功配置并且启动之后，等待20秒左右，关闭程序，然后得到输出的结果。

**代码示例如下:**

**Spout**  
用于发送消息。

import java.util.Map;
        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;
    
        /** * * Title: TestSpout * Description: * 发送信息 * Version:1.0.0 * @author pancm * @date 2018年3月6日 */
        public class TestSpout extends BaseRichSpout{ 
    
            private static final long serialVersionUID = 225243592780939490L;
    
            private SpoutOutputCollector collector;
            private static final String field="word";
            private int count=1;
            private String[] message =  {
        "My nickname is xuwujing",
        "My blog address is http://www.panchengming.com/",
        "My interest is playing games"
        };
    
            /** * open()方法中是在ISpout接口中定义，在Spout组件初始化时被调用。 * 有三个参数: * 1.Storm配置的Map; * 2.topology中组件的信息; * 3.发射tuple的方法; */
            @Override
            public void open(Map map, TopologyContext arg1, SpoutOutputCollector collector) {
                System.out.println("open:"+map.get("test"));
                this.collector = collector;
            }
    
        /** * nextTuple()方法是Spout实现的核心。 * 也就是主要执行方法，用于输出信息,通过collector.emit方法发射。 */
            @Override
            public void nextTuple() {
    
                if(count<=message.length){
                    System.out.println("第"+count +"次开始发送数据...");
                    this.collector.emit(new Values(message[count-1]));
                }
                count++;
            }
    
    
            /** * declareOutputFields是在IComponent接口中定义，用于声明数据格式。 * 即输出的一个Tuple中，包含几个字段。 */
            @Override
            public void declareOutputFields(OutputFieldsDeclarer declarer) {
                System.out.println("定义格式...");
                declarer.declare(new Fields(field));
            }
    
            /** * 当一个Tuple处理成功时，会调用这个方法 */
            @Override
            public void ack(Object obj) {
                System.out.println("ack:"+obj);
            }
    
            /** * 当Topology停止时，会调用这个方法 */
            @Override
            public void close() {
                System.out.println("关闭...");
            }
    
            /** * 当一个Tuple处理失败时，会调用这个方法 */
            @Override
            public void fail(Object obj) {
                System.out.println("失败:"+obj);
            }
    
        }

**TestBolt**

用于分割单词。

import java.util.Map;
    
        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;
    
    
        /** * * Title: TestBolt * Description: * 对单词进行分割 * Version:1.0.0 * @author pancm * @date 2018年3月16日 */
        public class TestBolt extends BaseRichBolt{ 
    
            /** * */
            private static final long serialVersionUID = 4743224635827696343L;
    
            private OutputCollector collector;
    
            /** * 在Bolt启动前执行，提供Bolt启动环境配置的入口 * 一般对于不可序列化的对象进行实例化。 * 注:如果是可以序列化的对象，那么最好是使用构造函数。 */
            @Override
            public void prepare(Map map, TopologyContext arg1, OutputCollector collector) {
                System.out.println("prepare:"+map.get("test"));
                this.collector=collector;
            }
    
            /** * execute()方法是Bolt实现的核心。 * 也就是执行方法，每次Bolt从流接收一个订阅的tuple，都会调用这个方法。 */
            @Override
            public void execute(Tuple tuple) {
                String msg=tuple.getStringByField("word");
            System.out.println("开始分割单词:"+msg);
        String[] words = msg.toLowerCase().split(" ");
        for (String word : words) {
        this.collector.emit(new Values(word));//向下一个bolt发射数据
        } 
    
            }
    
            /** * 声明数据格式 */
            @Override
            public void declareOutputFields(OutputFieldsDeclarer declarer) {
                declarer.declare(new Fields("count"));
            }
    
            /** * cleanup是IBolt接口中定义,用于释放bolt占用的资源。 * Storm在终止一个bolt之前会调用这个方法。 */
            @Override
            public void cleanup() {
                System.out.println("TestBolt的资源释放");
            }
        }

**Test2Bolt**  
用于统计单词出现次数。

import java.util.HashMap;
        import java.util.Map;
    
        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;
    
        /** * * Title: Test2Bolt * Description: * 统计单词出现的次数 * Version:1.0.0 * @author pancm * @date 2018年3月16日 */
        public class Test2Bolt extends BaseRichBolt{ 
    
            /** * */
            private static final long serialVersionUID = 4743224635827696343L;
    
    
            /** * 保存单词和对应的计数 */
            private HashMap<String, Integer> counts = null;
    
            private long count=1;
            /** * 在Bolt启动前执行，提供Bolt启动环境配置的入口 * 一般对于不可序列化的对象进行实例化。 * 注:如果是可以序列化的对象，那么最好是使用构造函数。 */
            @Override
            public void prepare(Map map, TopologyContext arg1, OutputCollector collector) {
                System.out.println("prepare:"+map.get("test"));
                this.counts=new HashMap<String, Integer>();
            }
    
            /** * execute()方法是Bolt实现的核心。 * 也就是执行方法，每次Bolt从流接收一个订阅的tuple，都会调用这个方法。 * */
            @Override
            public void execute(Tuple tuple) {
                String msg=tuple.getStringByField("count");
                System.out.println("第"+count+"次统计单词出现的次数");
                /** * 如果不包含该单词，说明在该map是第一次出现 * 否则进行加1 */
                if (!counts.containsKey(msg)) {
                    counts.put(msg, 1);
                } else {
                    counts.put(msg, counts.get(msg)+1);
                }
                count++;
            }
    
    
            /** * cleanup是IBolt接口中定义,用于释放bolt占用的资源。 * Storm在终止一个bolt之前会调用这个方法。 */
            @Override
            public void cleanup() {
                System.out.println("===========开始显示单词数量============");
                for (Map.Entry<String, Integer> entry : counts.entrySet()) {
                    System.out.println(entry.getKey() + ": " + entry.getValue());
                }
                System.out.println("===========结束============");
               System.out.println("Test2Bolt的资源释放");
            }
    
            /** * 声明数据格式 */
            @Override
            public void declareOutputFields(OutputFieldsDeclarer arg0) {
    
            }
        }

**Topology**

主程序入口。

import org.apache.storm.Config;
        import org.apache.storm.LocalCluster;
        import org.apache.storm.StormSubmitter;
        import org.apache.storm.topology.TopologyBuilder;
        import org.apache.storm.tuple.Fields;
    
        /** * * Title: App * Description: * storm测试 * Version:1.0.0 * @author pancm * @date 2018年3月6日 */
        public class App { 
    
            private static final String test_spout="test_spout"; 
            private static final String test_bolt="test_bolt"; 
            private static final String test2_bolt="test2_bolt"; 
    
            public static void main(String[] args)  {
                //定义一个拓扑
                TopologyBuilder builder=new TopologyBuilder();
                //设置一个Executeor(线程)，默认一个
                builder.setSpout(test_spout, new TestSpout(),1);
                //shuffleGrouping:表示是随机分组
                //设置一个Executeor(线程)，和一个task
                builder.setBolt(test_bolt, new TestBolt(),1).setNumTasks(1).shuffleGrouping(test_spout);
                //fieldsGrouping:表示是按字段分组
                //设置一个Executeor(线程)，和一个task
                builder.setBolt(test2_bolt, new Test2Bolt(),1).setNumTasks(1).fieldsGrouping(test_bolt, new Fields("count"));
                Config conf = new Config();
                conf.put("test", "test");
                try{
                  //运行拓扑
               if(args !=null&&args.length>0){ //有参数时，表示向集群提交作业，并把第一个参数当做topology名称
                 System.out.println("运行远程模式");
                     StormSubmitter.submitTopology(args[0], conf, builder.createTopology());
              } else{
       //没有参数时，本地提交
            //启动本地模式
                System.out.println("运行本地模式");
            LocalCluster cluster = new LocalCluster();
            cluster.submitTopology("Word-counts" ,conf,  builder.createTopology() );
            Thread.sleep(20000);
        // //关闭本地集群
            cluster.shutdown();
              }
                }catch (Exception e){
                    e.printStackTrace();
                }
            }
        }

输出结果:

运行本地模式
    定义格式...
    open:test
    第1次开始发送数据...
    第2次开始发送数据...
    第3次开始发送数据...
    prepare:test
    prepare:test
    开始分割单词:My nickname is xuwujing
    开始分割单词:My blog address is http://www.panchengming.com/
    开始分割单词:My interest is playing games
    第1次统计单词出现的次数
    第2次统计单词出现的次数
    第3次统计单词出现的次数
    第4次统计单词出现的次数
    第5次统计单词出现的次数
    第6次统计单词出现的次数
    第7次统计单词出现的次数
    第8次统计单词出现的次数
    第9次统计单词出现的次数
    第10次统计单词出现的次数
    第11次统计单词出现的次数
    第12次统计单词出现的次数
    第13次统计单词出现的次数
    第14次统计单词出现的次数
    ===========开始显示单词数量============
    address: 1
    interest: 1
    nickname: 1
    games: 1
    is: 3
    xuwujing: 1
    playing: 1
    my: 3
    blog: 1
    http://www.panchengming.com/: 1
    ===========结束============
    Test2Bolt的资源释放
    TestBolt的资源释放
    关闭...

上述的是本地模式运行，如果想在Storm集群中进行使用，只需要将程序打包为jar，然后将程序上传到storm集群中，  
输入:

**storm jar xxx.jar xxx xxx**  
**说明:第一个xxx是storm程序打包的包名,第二个xxx是运行主程序的路径，第三个xxx则表示主程序输入的参数，这个可以随意。**

如果是使用maven打包的话，则需要在pom.xml加上

<plugin>
              <artifactId>maven-assembly-plugin</artifactId>
              <configuration>
                <descriptorRefs>
                  <descriptorRef>jar-with-dependencies</descriptorRef>
                </descriptorRefs>
                <archive>
                  <manifest>
                    <mainClass>com.pancm.storm.App</mainClass>
                  </manifest>
                </archive>
              </configuration>
          </plugin>

成功运行程序之后，可以在Storm集群的UI界面查看该程序的状态。

到此，本文结束，谢谢阅读！

[Image 1]: 
[http_www.tianshouzhi.com_api_tutorials_storm_52]: http://www.tianshouzhi.com/api/tutorials/storm/52
[http_www.panchengming.com_2018_01_26_pancm70]: http://www.panchengming.com/2018/01/26/pancm70/