Storm的优点有哪些

本篇内容介绍了“Storm的优点有哪些”的有关知识，在实际案例的操作过程中，不少人都会遇到这样的困境，接下来就让小编带领大家学习一下如何处理这些情况吧！希望大家仔细阅读，能够学有所成！

创新互联公司专注于湘潭县网站建设服务及定制，我们拥有丰富的企业做网站经验。 热诚为您提供湘潭县营销型网站建设，湘潭县网站制作、湘潭县网页设计、湘潭县网站官网定制、小程序开发服务，打造湘潭县网络公司原创品牌,更为您提供湘潭县网站排名全网营销落地服务。

Storm和hadoop的区别

• 数据来源：HADOOP是HDFS上某个文件夹下的可能是成TB的数据，STORM是实时新增的某一笔数据

• 处理过程：HADOOP是分MAP阶段到REDUCE阶段，STORM是由用户定义处理流程，流程中可以包含多个步骤，每个步骤可以是数据源(SPOUT)或处理逻辑(BOLT)

• 是否结束：HADOOP最后是要结束的，STORM是没有结束状态，到最后一步时，就停在那，直到有新数据进入时再从头开始

• 处理速度：HADOOP是以处理HDFS上大量数据为目的，速度慢，STORM是只要处理新增的某一笔数据即可可以做到很快。

• 适用场景：HADOOP是在要处理一批数据时用的，不讲究时效性，要处理就提交一个JOB，STORM是要处理某一新增数据时用的，要讲时效性

• 与MQ对比：HADOOP没有对比性，STORM可以看作是有N个步骤，每个步骤处理完就向下一个MQ发送消息，监听这个MQ的消费者继续处理

好的编程模型让开发者专注于业务逻辑;不好的编程模型让开发者把时间花费在通信，处理异常等琐事上.

编程模型例子：

• 用hadoop的MapReduce和MPI做一个对比，在hadoop的MapReduce里面呢，它的编程模型里面呢，map和reduce，你只用去写map和reduce函数，以及一些简单的驱动，程序就能跑起来，你不用关心map和数据是怎么切分的，map和reduce是怎么传输的，reduce的数据是怎么写到hadoop的HDFS里面的，这些你都不用关心，看起来写mapreduce就是单机的代码，没有什么分布式的特点在里面啊，但是它运行的分布式框架来帮你做上述这些东西。

• 反过来我们看写MPI的程序就完全不一样，写MPI的时候你就会很明显的感觉到你在写一个并行分布式程序，你需要在很多地方显式的去调数据传输的接口，你还要显式的去调一些数据同步的接口，这样才能把MPI程序显式的给RUN起来，这就是编程模型不同导致的不同的开发体验，其实这不仅仅是开发容易不容易的问题，更主要的是一个开发效率的问题，其实更简单的程序更能写出健壮的程序，写mapreduce程序是很简单的，但是要写出一个稳定靠谱的MPI程序就难一些

架构

• Nimbus

• Supervisor

• Worker

编程模型：

• DAG

• Spout

• Bolt

数据传输：

• Zmq

• Zmq也是开源的消息传递的框架，虽然叫mq，但它并不是一个message queue，而是一个封装的比较好的

• Netty

• netty是NIO的网络框架，效率比较高。之所以有netty是storm在apache之后呢，zmq的license和storm的license不兼容的，bolt处理完消息后会告诉Spout。

高可用性

• 异常处理

• 消息可靠性保证机制

可维护性：

• Storm有个UI可以看跑在上面的程序监控

实时请求应答服务（同步），

• 实时请求应答服务（同步），往往不是一个很简单的操作，而且大量的操作，用DAG模型来提高请求处理速度

• DRPC

• 实时请求处理

• 例子：发送图片，或者图片地址，进行图片特征的提取

• 

• 这里DRPC Server的好处是什么呢？这样看起来就像是一个Server，经过Spout，然后经过Bolt，不是更麻烦了吗？DRPC Server其实适用于分布式，可以应用分布式处理这个单个请求，来加速处理的过程。

  DRPCClientclient = new DRPCClient("drpc-host", 3772);
  String result = client.execute("reach","http://twitter.com");
  //服务端由四部分组成：包括一个DRPC Server，一个DPRC Spout，一个Topology和一个ReturnResult。

  

流式处理（异步）--- 不是说不快，而是不是等待结果

• 逐条处理

• 例子：ETL，把关心的数据提取，标准格式入库，它的特点是我把数据给你了，不用再返回给我，这个是异步的

• 分析统计

• 例子：日志PV，UV统计，访问热点统计，这类数据之间是有关联的，比如按某些字段做聚合，加和，平均等等

• 

• 最后写到redis，Hbase，MySQL，或者其他的MQ里面去给其他的系统去消费。

  /**
   * ShuffleGrouping("spout")就是从spout来订阅数据，fieldGrouping("split", new Fields("word"))
   * 实际上就是一个hash，同一个词有相同的hash，然后就会被hash到同一个WordCount的bolt里面，然后就
   * 可以进行计数。接下来两行呢是配置文件，然后是配置3个worker，接下来是通过Submitter提交Topology
   * 到Storm集群里面去。程序会编译打包，这段代码来自storm里面的starter的一段代码，这个代码怎么真正
   * 运行起来呢，就用storm jar 然后jar包的名，然后就是类的名字，和topology的名字，因为这里有个args[0]。
   * 这段代码很简单，首先呢，第一部分构造了一个DAG的有向无环图，然后生成配置，提交到Storm集群去。
   * */
  public static void main(String[] args) {
          TopologyBuilder builder = new TopologyBuilder();
          builder.setSpout("spout", new RandomSentenceSpout(), 5);
          builder.setBolt("split", new SplitSentence(), 8).shuffleGrouping("spout");
          builder.setBolt("count", new WordCount(), 12).fieldsGrouping("split", new Fields("word"));
  
          Config conf = new Config();
          conf.setDebug(true);
          
          if(args != null && args.length > 0) {
              conf.setNumWorkers(3);
  
              StormSubmitter.submitTopologyWithProgressBar(args[0], conf, builder.createTopology());
          }
      }
  
  ###Linux: storm jar storm-starter-0.9.4.jar storm.starter.WordCountTopology wordcount

   

Cluster Summary（整个集群的）

• 一个slot就是一个worker，一个worker里面是一个jvm，一个worker里面呢可以有多个executor，每一个executor就是执行线程，每一个executor上面执行一个或多个Task，一般来说默认是一个task。

• Topology Summary（每个应用程序的）

• 一个应用程序就是一个Topology，它有名字，还有ID，然后有个状态，ACTIVE就是正在运行，KILLED就是已经被杀掉了。

• 

• Topology actions就是可以对Topology采取一些操作，Deactivate就是暂停，Rebalance就是重新做一下balance，然后kill就是杀掉这个应用。

• 这个应用运行的到底怎么样呢，在Topology stats里面有个整体的统计，有10分钟，3小时，1天，还有所有的统计，这里面比较关键的呢，是Complete latency，它的意思就是一条数据从发出去到处理完花了多长时间，第二个比较关键的呢就是ACK，这个反映的是吞吐，前面的Complete latency反映的延迟。

• 在Spouts的统计信息里面呢，一个是spout的名字，和代码里面是对应的，第二个呢是这个spout它有多少个executor，然后呢它有多少个task，然后呢是它在一定时间内往外emit出多少数据，真正tranfer传输了多少数据，然后它latency延迟是多少，然后ACK处理了多少数据，后面还有错误的信息。

• Bolt也类似，通过这个UI页面可以实时观看这些统计信息，是非常有用的，可以知道哪个环节比较慢，哪些地方有没有什么瓶颈了，有瓶颈了是不是加一个并发来解决问题。

• Spout中这里最关键的是一个nextTuple()，它是从外部取数据的源头，可以从DPRC取数据，可以从MQ，比如Kafka中取数据，然后给后面的bolt进行处理，然后这里wordcount没有那么复杂，就自己随机的生成了数据。

• _collector.emit(new Values(sentence), new Object());

• 这个代码后面new Object()等于是随机的生成了一个message的ID，这个ID有什么用，后面会讲到，实际上它是消息可靠性保障的一部分。有了这个ID呢Storm就可以帮你去跟踪这条消息到底有没有被处理完，如果处理完了呢?

• 如果处理完了，它就是调用一个ack告诉spout，我已经处理完了，这里ack方法里面仅仅是把id打印出来，因为这里id没有什么意义，仅仅是为了展示，相反，如果在一定时间内没有处理完，会调用fail告诉说消息处理失败了。

  /**
   * Licensed to the Apache Software Foundation (ASF) under one
   * or more contributor license agreements.  See the NOTICE file
   * distributed with this work for additional information
   * regarding copyright ownership.  The ASF licenses this file
   * to you under the Apache License, Version 2.0 (the
   * "License"); you may not use this file except in compliance
   * with the License.  You may obtain a copy of the License at
   *
   * http://www.apache.org/licenses/LICENSE-2.0
   *
   * Unless required by applicable law or agreed to in writing, software
   * distributed under the License is distributed on an "AS IS" BASIS,
   * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
   * See the License for the specific language governing permissions and
   * limitations under the License.
   */
  package storm.starter.spout;
  
  import backtype.storm.spout.SpoutOutputCollector;
  import backtype.storm.task.TopologyContext;
  import backtype.storm.topology.OutputFieldsDeclarer;
  import backtype.storm.topology.base.BaseRichSpout;
  import backtype.storm.tuple.Fields;
  import backtype.storm.tuple.Values;
  import backtype.storm.utils.Utils;
  
  import java.util.Map;
  import java.util.Random;
  
  public class RandomSentenceSpout extends BaseRichSpout {
    SpoutOutputCollector _collector;
    Random _rand;
  
  
    @Override
    public void open(Map conf, TopologyContext context, SpoutOutputCollector collector) {
      _collector = collector;
      _rand = new Random();
    }
  
  
    @Override
    public void nextTuple() {
      Utils.sleep(100);
      String[] sentences = new String[]{ "the cow jumped over the moon", "an apple a day keeps the doctor away",
          "four score and seven years ago", "snow white and the seven dwarfs", "i am at two with nature" };
      String sentence = sentences[_rand.nextInt(sentences.length)];
      _collector.emit(new Values(sentence), new Object());
    }
  
    @Override
    public void ack(Object id) {
  	  System.out.println(id);
    }
  
    @Override
    public void fail(Object id) {
  	  System.out.println(id);
    }
  
    @Override
    public void declareOutputFields(OutputFieldsDeclarer declarer) {
      declarer.declare(new Fields("word"));
    }
    @Override
    public  Map getComponentConfiguration(){
      return null;
  
    }
  }
  
  public static  class WordCount1 extends BaseBasicBolt{
          Map counts = new HashMap();
          @Override
          public void execute(Tuple tuple, BasicOutputCollector basicOutputCollector) {
              String word = tuple.getStringByField("word");
              Integer count = counts.get(word);
              if(count==null){
                  count=0;
              }
              count++;
              counts.put(word,count);
              System.out.println("word++"+word+"========="+count);
              basicOutputCollector.emit(new Values(word,count));
          }
  
          @Override
          public void declareOutputFields(OutputFieldsDeclarer outputFieldsDeclarer) {
              outputFieldsDeclarer.declare(new Fields("word","count"));
          }
      }

   

• 对于wordcount的示例，它是有两个blot，一个bolt是分词，一个bolt是计数，这里SplitSentence是展示它支持多语言的开发，其实这里代码调用的是python的splitsentence.py，使用的是ShellBolt这个组件

• 那wordcount这个bolt是用java实现的，它的实现核心是亮点，一点是有execute这样一个函数，第二个是declareOutputFields这个函数，这两个函数的作用其实是很什么呢？最核心的其实是execute，execute的作用呢就是拿到输入的数据Tuple，然后再emit数据出去。

• 以上就是在storm里面一个最简单的wordcount的例子，它的主函数的代码，它的提交的命令行代码，Spout是什么样的，Bolt是什么样的，提交到Storm集群之后是一个什么样的运行状况，在WebUI上面看到哪些核心的信息，这个在后面的应用开发里面都会大量的运用到。

Storm与其他技术对比

• Storm：进程、线程常驻运行，数据不进入磁盘，网络传递。

• MapReduce：TB、PB级别数据设计的，一次的批处理作业。

• Storm：纯流式处理，处理数据单元是一个个Tuple。另外Storm专门为流式处理设计，它的数据传输模式更为简单，很多地方也更为高效。并不是不能做批处理，它也可以来做微批处理，来提高吞吐。

• Spark Streaming：微批处理，一个批处理怎么做流式处理呢，它基于内存和DAG可以把处理任务做的很快，把RDD做的很小来用小的批处理来接近流式处理。

• 和其它如MPI系统相比

通过对比，更能了解Storm的一些特点：

• 首先，相对于Queue+Worker来说，它是一个通用的分布式系统，分布式系统的一些细节可以屏蔽掉，比如说水平扩展，容错，上层应用只需要关注自己的业务逻辑就可以了，这一点对应应用开发人员来说是非常重要的，不然的话业务逻辑会被底层的一些细节所打乱。

• 另外一个，Storm作为一个纯的流式处理系统，和mapreduce的差异相当大，一种称为流式处理，一种称为批处理，Storm是一个常驻运行的，它的消息收发是很高效的。

• 和spark这种微批处理系统相比呢，Storm可以处理单条单条的消息。

• 总的来说呢，Storm在设计之初呢，就被定义为分布式的流式处理系统，所以说大部分的流式计算需求都可以通过Storm很好的满足，Storm目前在稳定性方面也做的相当不错，对于实时流式计算来说是个非常不错的选择

“Storm的优点有哪些”的内容就介绍到这里了，感谢大家的阅读。如果想了解更多行业相关的知识可以关注创新互联网站，小编将为大家输出更多高质量的实用文章！