必威体育Betway必威体育官网
当前位置:首页 > IT技术

Storm实时流计算原理概述与最佳入门实践

时间:2019-09-07 10:12:13来源:IT技术作者:seo实验室小编阅读:83次「手机版」
 

流计算

随着互联网的发展,信息量爆炸式的增长,人们越来越需要实时获取一些计算信息,离线计算已经不能满足了人们的需求,这时Storm、Flink、Spark Streaming等实时计算框架日益发展起来。本篇文章主要讲述Storm原理架构概述以及入门实践案例的编写。

一、Storm架构原理概述

1.Storm的优点

  1. Storm是一款开源免费的分布式,可容错性,可扩展、高可靠的实时流处理框架,它可以实时处理无界的流数据,并且支持多种编程语言的开发
  2. Storm涉及领域广泛,比如实时数据分析机器学习,持续计算,分布式RPC,ETL数据处理等方面。

2. Storm核心组件

这里写图片描述

Storm 主要有以下概念:

  • Spout 产生数据源的地方。
  • Bolt 消息处理者。
  • Topology 网络拓扑。
  • Stream 流。
  • Tuple 元组。
  • Stream 消息流。
  • Streaming Group消息流组。

Spout消息源

正如上图所示,Spout消息源就像一个水龙头,源源不断的产生数据,ISpout是一个接口,主要有以下方法:

public interface ISpout extends serializable {
    //实例化Spout,主要是一些集群配置
    void open(Map conf, Topologycontext context, SpoutOutputCollector collector);
    //关闭Spout,Storm并不保证关闭,可以通过kill -9的方式显示杀死
    void close();

    void activate();

    void deactivate();
    //发射下一个tuple
    void nextTuple();
    //Spout发送每一个tuple会有一个msgId作为表示,如果tuple被完全处理,则返回ack消息
    void ack(Object msgId);
    //如果发射出去的tuple没有被完全处理,返回fail消息,隔一段时间重复发送
    void fail(Object msgId);
}

在我们的应用程序中编写Spout需要实现ISpout接口,一般情况下,我们并不会直接实现ISpout接口,而是集成它的抽象类BaseRichSpout,该抽象类覆盖了ISpout接口的这几个方法,继承BaseRichSpout抽象类的好处是我们可以自由的按照我们的需求编写程序,而不需实现全部方法,ISpout接口有很多实现类和子接口:

这里写图片描述

BaseRichSpout抽象类实现了ISpout接口,集成了BaseCompont接口,而BaseCompont实现了Icomponent接口,因此,该抽象类总共实现了ISpout接口和IComponent接口,实现了更加丰富的功能,源码如下所示:

public abstract class BaseRichSpout extends BaseComponent implements IRichSpout {
    @Override
    public void close() {
    }

    @Override
    public void activate() {
    }

    @Override
    public void deactivate() {
    }

    @Override
    public void ack(Object msgId) {
    }

    @Override
    public void fail(Object msgId) {
    }
}

Bolt消息处理者

Bolt以Tuple元组数据作为输入数据,经过处理之后产生一个新的数据输出。IBolt是一个接口,实现了和ISpout类似的方法,源码如下:

public interface IBolt extends Serializable {
   //初始化bolt
    void prepare(Map stormConf, TopologyContext context, OutputCollector collector);
   //实现逻辑处理
    void execute(Tuple input);

    void cleanup();
}

同样,我们在编写应用程序的时候集成BaseRichBolt抽象类,它实现了IBolt接口和IComponent接口,其源码如下所示:

public abstract class BaseRichBolt extends BaseComponent implements IRichBolt {
    @Override
    public void cleanup() {
    }    
}

Topology网络拓扑

如果说Stream流数据,Spout,Bolt是流计算的血肉,那么Topology就是流计算的骨架,它将所有的Spout和Bolt组织在一起,构成了一套实时流处理架构。

Topology的运行有两种模式,即Storm运行有两种模式,本地模式和集群模式。

本地模式,主要用于我们开发测试应用程序来用,可以通过 localcluster来创建一个集群,例如使用如下代码创建本地模式提交任务:

LocalCluster localCluster=new LocalCluster();

localCluster.submitTopology("LocalSumTopology",new Config(),builder.createTopology());

集群模式,用于生产线上使用,我们可以定义一个Topology,然后使用 StormSubmitter来提交任务,如下例所示:

Config conf = new Config();
conf.setNumWorkers(20);
conf.setMaxSpoutPending(5000);
StormSubmitter.submitTopology("mytopology", conf, topology);

Stream消息流

它是一个抽象的概念,是一个无界的 Tuple 序列,这些 Tuple 以分布式的方式来并行的创建和处理。

Tuple元组

Tuple是Storm中最基本的数据处理单元,是一个值列表,可以支持任何类型的数据,如果自定义类型的话需要进行序列化。

Stream Grouping消息流组

这里写图片描述

如上图所示,Stream Grouping主要是定义Bolt处理由Spout发射的何种数据,比如按照某个字段分组等等,主要有以下分组策略:

  • 随机分组(Shuffle Grouping):随机分发到Bolt的各个任务中,保证每个任务获取相同数量的元组。
  • 字段分组(fields Grouping):根据指定 字段分割数据流并分组。
  • 局部Key分组(Partial Key Grouping):和Feilds Grouping一样指定Key发送数据,但是它会将数据负载均衡的发送给下游的数据,可以提供更好的资源利用率当发生数据倾斜的时候
  • 全部分组(All Grouping):对于每一个Tuple,所有的Bolt都会收到。
  • 全局分组(Global Grouping):全部的流分配到一个Bolt的同一个任务中,即分配给ID最小的Task。
  • 无分组(None Grouping):等效于随机分组。
  • 直接分组(Direct Grouping):即元组生产者决定有哪一个消费者消费数据,Spout必须使用emitDirect方法直接发射。
  • 本地/随机分组(Local or shuffle grouping):如果目标Bolt有一个或者多个task在相同的Worker进程中,task则会只发送到处理这些task的Worker进程中,否则和Shuffle Grouping一样随机分发。

3.strom集群架构

这里写图片描述

Strom使用Zookeeper作为协调框架,主要由nimbus,supervisor,worker等组件组成。它们三者之间主要有以下关系:

nimbus为主节点,supervisor为从节点,在Surpervisor中可以启动多个Worker进程,每一个Wroker进程为一个特定的Topology服务,一个Topology可以建立在多个Wroker进程之上,在一个Worker中可以启动多个Executor线程,在executor中可以启动多个task任务。一般情况下,Executor要么全部是Spout的task,要么全部是Bolt的task,也就是说一个Executor只能运行一个Spout或者Bolt的task任务。

二、简单案例编写

下面主要编写一个求和案例,Spout端不断的输入数据,从1,2,3,,,n,Bolt端接收数据进行累加计算,最后在控制台打印,首先导入maven依赖:

<dependency>
      <groupId>org.apache.storm</groupId>
      <artifactId>storm-core</artifactId>
      <version>${storm.version}</version>
    </dependency>

    <dependency>
      <groupId>commons-io</groupId>
      <artifactId>commons-io</artifactId>
      <version>2.4</version>
    </dependency>

程序代码如下:

/**
 * 实现简单的本地求和功能
 */
public class LocalSumTopology {


    /**
     * Spout组件
     * 产生数据并且发送
     */
    public static class SumSpout extends BaseRichSpout{

        private SpoutOutputCollector collector;

        /**
         * 初始化操作
         * @param conf 初始化配置项
         * @param context 上下文
         * @param collector 数据发射器
         */
        @Override
        public void open(Map conf, TopologyContext context, SpoutOutputCollector collector) {
            this.collector=collector;
        }

        int number=0;
        /**
         * 发射数据
         * 该方法是一个死循环
         */
        @Override
        public void nextTuple() {
            //Values类实现了ArrayList
            collector.emit(new Values(++number));

            System.out.println("Spout number: "+number);
            //防止数据产生太快,睡眠一秒
            Utils.sleep(1000);

        }

        /**
         * 定义输出端字段
         * @param declarer
         */
        @Override
        public void declareOutputFields(OutputFieldsDeclarer declarer) {
            //与上面的number变量对应
            declarer.declare(new Fields("num"));
        }
    }

    /**
     * Bolt组件
     * 实现业务的逻辑处理,这里求和
     */
    public static class SumBolt extends BaseRichBolt{
        /**
         * 因为 这里接收数据之后不需要再发送给下一个Bolt,因此再初始化collector发射器
         * @param stormConf
         * @param context
         * @param collector
         */
        @Override
        public void prepare(Map stormConf, TopologyContext context, OutputCollector collector) {

        }

        int sum=0;
        /**
         * 执行业务逻辑的处理
         * 该方法也是一个死循环
         * @param input
         */
        @Override
        public void execute(Tuple input) {
            //可以通过字段名或者下标索引获取
            integer value=input.getIntegerByField("num");
            sum+=value;
            System.out.println("Bolt sum: "+sum);
        }

        @Override
        public void declareOutputFields(OutputFieldsDeclarer declarer) {

        }
    }


    public static void main(String[] args) {
        //使用TopologyBuilder设置Spout和Bolt,并且将其关联 在一起
        //创建Topology
        TopologyBuilder builder=new TopologyBuilder();
        builder.setSpout("SumSpout",new SumSpout());
        builder.setBolt("SumBolt",new SumBolt()).shuffleGrouping("SumSpout");
        //使用本地模式
        LocalCluster localCluster=new LocalCluster();
        localCluster.submitTopology("LocalSumTopology",new Config(),builder.createTopology());
    }


}

如果想在集群中提交该应用程序,只需要将程序中的本地模式使用StormSubmitter改为集群模式,使用下面命令提交即可:

storm jar test.jar main.java args

相关阅读

webstorm 2019 激活破解方法大全

webstorm 2019 激活破解方法大全 方法一:使用破解补丁 这个一劳永逸推荐用这个激活。 最新版本下载地址:IntelliJ IDEA 破解补丁

Android OTA升级原理和流程分析(一) --update.zip包的制

Android OTA升级原理和流程分析(一)–update.zip包的制作   这篇及以后的篇幅将通过分析update.zip包在具体Android系统升级的过

SPF 记录:原理、语法及配置方法简介

摘自网址:https://www.renfei.org/blog/introduction-to-spf.html点击打开链接很大部分是摘下来滴,文章很棒那,可以看原版链接,侵删

电容器原理和计算公式及电容单位换算

电容器原理在生产及生活实践中,科学家们发现,凡是被绝缘隔开的两个导体之间加以电压时,则接在高电位的导体就能容纳正电荷,接在低电位

ViBe算法原理详解

Vibe算法是一种基于背景更新的前景检测算法,其原理是通过提取像素点(x, y)周围的像素值及以前的像素值建立像素点的样本集,然后再将另

分享到:

栏目导航

推荐阅读

热门阅读