Storm搭建及介绍（9）

女爷i 2022-05-29 06:11 152阅读 0赞

# 一、Storm到底是什么？ #

## 1、mysql，hadoop与storm ##

mysql、hadoop与storm的关系  
mysql：事务性系统，面临海量数据的尴尬  
hadoop：离线批处理  
storm：实时计算  
**mysql、hadoop与storm的关系**  
![这里写图片描述][Image 1]

# 2、我们能不能自己搞一套storm？ #

来一条数据，我理解就算一条，来一条，算一条

坑，海量高并发大数据，高并发的请求数据，分布式的系统，流式处理的分布式系统

如果自己搞一套实时流系统出来，也是可以的，但是。。。。

（1）花费大量的时间在底层技术细节上：如何部署各种中间队列，节点间的通信，容错，资源调配，计算节点的迁移和部署，等等

（2）花费大量的时间在系统的高可用上问题上：如何保证各种节点能够高可用稳定运行

（3）花费大量的时间在系统扩容上：吞吐量需要扩容的时候，你需要花费大量的时间去增加节点，修改配置，测试，等等

5万/s，10万/s，扩容

国内，国产的实时大数据计算系统，唯一做出来的，做得好的，做得影响力特别大，特别牛逼的，就是JStorm，阿里

阿里，技术实力，世界一流，顶尖，国内顶尖，一流

JStorm，clojure编程预压，Java重新写了一遍，Galaxy流式计算的系统，百度，腾讯，也都自己做了，也能做得很好

# 3、storm的特点是什么？ #

（1）支撑各种实时类的项目场景：实时处理消息以及更新数据库，基于最基础的实时计算语义和API（实时数据处理领域）；对实时的数据流持续的进行查询或计算，同时将最新的计算结果持续的推送给客户端展示，同样基于最基础的实时计算语义和API（实时数据分析领域）；对耗时的查询进行并行化，基于DRPC，即分布式RPC调用，单表30天数据，并行化，每个进程查询一天数据，最后组装结果

storm做各种实时类的项目都ok

（2）高度的可伸缩性：如果要扩容，直接加机器，调整storm计算作业的并行度就可以了，storm会自动部署更多的进程和线程到其他的机器上去，无缝快速扩容

扩容起来，超方便

（3）数据不丢失的保证：storm的消息可靠机制开启后，可以保证一条数据都不丢

数据不丢失，也不重复计算

（4）超强的健壮性：从历史经验来看，storm比hadoop、spark等大数据类系统，健壮的多的多，因为元数据全部放zookeeper，不在内存中，随便挂都不要紧

特别的健壮，稳定性和可用性很高

（5）使用的便捷性：核心语义非常的简单，开发起来效率很高

用起来很简单，开发API还是很简单的

# 二、Storm的集群架构以及核心概念 #

## 1、Storm的集群架构 ##

Nimbus，Supervisor，ZooKeeper，Worker，Executor，Task  
![这里写图片描述][Image 1]

## 2、Storm的核心概念 ##

Topology，Spout，Bolt，Tuple，Stream  
![这里写图片描述][Image 1]  
拓扑：务虚的一个概念

Spout：数据源的一个代码组件，就是我们可以实现一个spout接口，写一个java类，在这个spout代码中，我们可以自己尝试去数据源获取数据，比如说从kafka中消费数据

bolt：一个业务处理的代码组件，spout会将数据传送给bolt，各种bolt还可以串联成一个计算链条，java类实现了一个bolt接口

一堆spout+bolt，就会组成一个topology，就是一个拓扑，实时计算作业，spout+bolt，一个拓扑涵盖数据源获取/生产+数据处理的所有的代码逻辑，topology

tuple：就是一条数据，每条数据都会被封装在tuple中，在多个spout和bolt之间传递

stream：就是一个流，务虚的一个概念，抽象的概念，源源不断过来的tuple，就组成了一条数据流

# 三、Storm的并行度以及流分组 #

**并行度和流分组**  
![这里写图片描述][Image 1]  
并行度：Worker->Executor->Task，没错，是Task

流分组：Task与Task之间的数据流向关系

Shuffle Grouping：随机发射，负载均衡  
Fields Grouping：根据某一个，或者某些个，fields，进行分组，那一个或者多个fields如果值完全相同的话，那么这些tuple，就会发送给下游bolt的其中固定的一个task

你发射的每条数据是一个tuple，每个tuple中有多个field作为字段

比如tuple，3个字段，name，age，salary

\{“name”: “tom”, “age”: 25, “salary”: 10000\} -> tuple -> 3个field，name，age，salary

All Grouping  
Global Grouping  
None Grouping  
Direct Grouping  
Local or Shuffle Grouping

# 四、例子 #

大数据，入门程序，wordcount，单词计数

你可以认为，storm源源不断的接收到一些句子，然后你需要实时的统计出句子中每个单词的出现次数

（1）搭建工程环境

<?xml version="1.0" encoding="UTF-8"?>
    <project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
    
      <modelVersion>4.0.0</modelVersion>
      <artifactId>storm-wordcount</artifactId>
      <packaging>jar</packaging>
    
      <name>storm-wordcount</name>
    
      <properties>
        <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
      </properties>
    
      <dependencies>
        <dependency>
          <groupId>org.apache.storm</groupId>
          <artifactId>storm-core</artifactId>
          <version>1.1.0</version>
        </dependency>
        <dependency>
          <groupId>commons-collections</groupId>
          <artifactId>commons-collections</artifactId>
          <version>3.2.1</version>
        </dependency>
        <dependency>
          <groupId>com.google.guava</groupId>
          <artifactId>guava</artifactId>
        </dependency>
      </dependencies>
    
      <build>
        <sourceDirectory>src/main/java</sourceDirectory>
        <testSourceDirectory>test/main/java</testSourceDirectory>
    
        <plugins>
            <plugin>
                <groupId>org.apache.maven.plugins</groupId>
                <artifactId>maven-shade-plugin</artifactId>
                <configuration>
                    <createDependencyReducedPom>true</createDependencyReducedPom>
                    <filters>
                        <filter>
                            <artifact>*:*</artifact>
                            <excludes>
                                <exclude>META-INF/*.SF</exclude>
                                <exclude>META-INF/*.sf</exclude>
                                <exclude>META-INF/*.DSA</exclude>
                                <exclude>META-INF/*.dsa</exclude>
                                <exclude>META-INF/*.RSA</exclude>
                                <exclude>META-INF/*.rsa</exclude>
                                <exclude>META-INF/*.EC</exclude>
                                <exclude>META-INF/*.ec</exclude>
                                <exclude>META-INF/MSFTSIG.SF</exclude>
                                <exclude>META-INF/MSFTSIG.RSA</exclude>
                            </excludes>
                        </filter>
                    </filters>
                </configuration>
                <executions>
                    <execution>
                        <phase>package</phase>
                        <goals>
                            <goal>shade</goal>
                        </goals>
                        <configuration>
                            <transformers>
                                <transformer implementation="org.apache.maven.plugins.shade.resource.ServicesResourceTransformer" />
                                <transformer implementation="org.apache.maven.plugins.shade.resource.ManifestResourceTransformer">
                                </transformer>
                            </transformers>
                        </configuration>
                    </execution>
                </executions>
            </plugin>
    
          <plugin>
            <groupId>org.codehaus.mojo</groupId>
            <artifactId>exec-maven-plugin</artifactId>
            <version>1.2.1</version>
            <executions>
              <execution>
                <goals>
                  <goal>exec</goal>
                </goals>
              </execution>
            </executions>
            <configuration>
              <executable>java</executable>
              <includeProjectDependencies>true</includeProjectDependencies>
              <includePluginDependencies>false</includePluginDependencies>
              <classpathScope>compile</classpathScope>
              <mainClass></mainClass>
            </configuration>
          </plugin>
        </plugins>
      </build>
    </project>

（2）编写代码

public class WordCountTopology { 
    
        public static 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[]{sentence("the cow jumped over the moon"), sentence("an apple a day keeps the doctor away"),
                    sentence("four score and seven years ago"), sentence("snow white and the seven dwarfs"), sentence("i am at two with nature")};
            final String sentence = sentences[_rand.nextInt(sentences.length)];
    
            _collector.emit(new Values(sentence));
          }
    
          protected String sentence(String input) {
            return input;
          }
    
          @Override
          public void ack(Object id) {
          }
    
          @Override
          public void fail(Object id) {
          }
    
          @Override
          public void declareOutputFields(OutputFieldsDeclarer declarer) {
            declarer.declare(new Fields("word"));
          }
    
        }
    
      public static class SplitSentence implements IRichBolt { 
    
        public SplitSentence() {
    
        }
    
        @Override
        public void declareOutputFields(OutputFieldsDeclarer declarer) {
          declarer.declare(new Fields("word"));
        }
    
        @Override
        public Map<String, Object> getComponentConfiguration() {
          return null;
        }
      }
    
      public static class WordCount extends BaseBasicBolt { 
        Map<String, Integer> counts = new HashMap<String, Integer>();
    
        @Override
        public void execute(Tuple tuple, BasicOutputCollector collector) {
          String word = tuple.getString(0);
          Integer count = counts.get(word);
          if (count == null)
            count = 0;
          count++;
          counts.put(word, count);
          collector.emit(new Values(word, count));
        }
    
        @Override
        public void declareOutputFields(OutputFieldsDeclarer declarer) {
          declarer.declare(new Fields("word", "count"));
        }
      }
    
      public static void main(String[] args) throws Exception {
        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());
        }
        else {
          conf.setMaxTaskParallelism(3);
    
          LocalCluster cluster = new LocalCluster();
          cluster.submitTopology("word-count", conf, builder.createTopology());
    
          Thread.sleep(10000);
    
          cluster.shutdown();
        }
      }
    }

（4）测试代码  
蕴含了很多的知识点

（1）Spout  
（2）Bolt  
（3）OutputCollector，Declarer  
（4）Topology  
（5）设置worker，executor，task，流分组

storm的核心基本原理，基本的开发，学会了

storm集群部署，怎么将storm的拓扑扔到storm集群上去跑

# 五、部署一个storm集群 #

（1）安装Java 7和Pythong 2.6.6

（2）下载storm安装包，解压缩，重命名，配置环境变量

（3）修改storm配置文件

mkdir /var/storm

conf/storm.yaml

storm.zookeeper.servers:  
\- “111.222.333.444”  
\- “555.666.777.888”

storm.local.dir: “/mnt/storm”