Storm入门（一）：编程模型

solve

发布于 2019-10-30 19:02:41

5570

文章被收录于专栏：大数据技术栈大数据技术栈

前言

本文是 storm 入门第一篇，因为 Storm 的本地模式体验极其简单，故而我希望第一篇我们先来体验一下 Storm，而不是其他分布式技术那样，开门就是架构，简介....

1 Storm初体验之本地运行

1.1 下载 Storm Jar 包

这里我们直接用 Maven 管理，直接在我们项目的 pom.xml 文件下加入：

   <!-- https://mvnrepository.com/artifact/org.apache.storm/storm-core -->
        <dependency>
            <groupId>org.apache.storm</groupId>
            <artifactId>storm-core</artifactId>
            <version>1.2.2</version>
        </dependency>

1.2 创建 Topology

什么是 Topology ？ Storm 为了方便编程，将 Storm 的程序封装成一个个的 Topology，这个Topology 也就是我们本文的重点编程模型。（其实质是一个 DAG 有向无环图）
Topology 是怎么样的？ Topology 包含以下几个结构：
1. 数据 Tuple：在 Storm 中，所有的数据都是以 Tuple 的形式进行传输的
2. 数据发送者 Spout：这个数据发送只是相对 Storm 本身来说的，Spout 一般从指定的外部数据源读取数据封装成 Tuple，进行数据的发送。
3. 数据处理组件 Bolt：Spout 的数据会发送到 Bolt，Bolt 就是用来做数据处理的组件，为了提高效率，一般 Bolt 只会处理一些单一的功能，然后会将数据继续往下一个 Bolt发送，形成一个 Bolt 链。
4. Stream Grouping：在数据从 Spout 到 Bolt 或者从 Bolt 到 Bolt 的时候可以指定数据的流向规则，这个规则就是 Stream Grouping。
5. Stream 数据流：从 Spout 发出，到 Bolt 处理完形成的数据通道就是一个数据流，一个Spout 可以发送多个数据流。
Topology 如何创建？好了，现在我们正式开始编码吧...
1. 创建 Spout：

 //继承自 BaseRichSpout 
static class MySpout extends BaseRichSpout {

      private Map map;
      private TopologyContext topologyContext;
      private SpoutOutputCollector spoutOutputCollector;

      //初始化函数
      public void open(Map map, TopologyContext topologyContext, SpoutOutputCollector spoutOutputCollector) {
          this.map = map;
          this.topologyContext = topologyContext;
          this.spoutOutputCollector = spoutOutputCollector;
      }

      //模拟的外部数据
      String[] outData = new String[]{"张三", "李四", "王五"};
      String[] outData2 = new String[]{"12", "13", "22"};

      /**
       * 我们可以在这里来模拟从外部获取数据并发送到 bolt
       * 该函数会在 storm 运行期间被循环调用
       */
      public void nextTuple() {
          String name = outData[(int) (Math.random() * 3)];
          String age = outData2[(int) (Math.random() * 3)];
          //将数据封装到 Tuple 里面
          Values v = new Values(name,age);
          // 将数据发送出去
          spoutOutputCollector.emit(v);

          //休眠一下，便于观察
          try {
              Thread.sleep(1000);
          } catch (InterruptedException e) {
              e.printStackTrace();
          }
      }

      /**
       * 数据声明
       * 发送的数据，通过这里的声明 告诉下游，我这个数据是什么
       * 相当于 表中的字段名
       */
      public void declareOutputFields(OutputFieldsDeclarer outputFieldsDeclarer) {
          //这里我们发送的是  姓名 和 年龄，主要顺序不能乱
          outputFieldsDeclarer.declare(new Fields("name","age"));
      }
  }

创建 Bolt

  //继承自 BaseRichBolt
  static class MyBolt extends BaseRichBolt {

      private Map stormConf;
      private TopologyContext context;
      private OutputCollector collector;
      //初始化函数
      public void prepare(Map stormConf, TopologyContext context, OutputCollector collector) {
          this.stormConf = stormConf;
          this.context = context;
          this.collector = collector;
      }
      // 处理数据  
      public void execute(Tuple input) {
          //获取上游发送的 name 字段
          String name = input.getStringByField("name");
          //获取上游发送的 age 字段
          String age = input.getStringByField("age");
          //这里我们简单的打印一下就好
          
          System.out.println(name + "****"+age);
          // 如果你还要继续往下发送 那么：collector.emit() 就可以
      }
      
      public void declareOutputFields(OutputFieldsDeclarer declarer) {
              //和 Spout 一样， 如果你还要继续往下发送数据，
              // 那么你就要在这里声明的发送的数据是什么
              // 我们这里不往下游发送，所以可以不用写
      }
  }

  //在主程序里面进行组装提交
  public static void main(String[] args) {
      // Topology 的构建者
      TopologyBuilder builder = new TopologyBuilder();
      // 设置 Spout ，并为其命名为 textSpout
      builder.setSpout("textSpout", new MySpout(), 3);
      //设置 Bolt，并为其命名为 MyBolt，
      builder.setBolt("MyBolt", new MyBolt(), 3)
      // 设置其 Stream Grouping 为 shuffleGrouping，并且是从 textSpout 接受数据
      .shuffleGrouping("textSpout");
      // 创建Topology
      StormTopology topology  = builder.createTopology()
      // 创建一个 本地集群
      LocalCluster localCluster = new LocalCluster();
      Config map = new Config();
      map.setNumAckers(1);
      //将 topology   提交到集群运行
      localCluster.submitTopology("test", map, topology  );
  }