storm

免费、开源、分布式、实时计算系统。 吞吐量高。 每秒每节点百万元组。

Spout //水龙头

Cloujr

JStorm

  1. storm                hadoop
  1. 实时流处理            批处理
  2. 无状态                有状态
  3. 使用zk协同的主        无zk的主从架构。
  4. 从架构
  6. 每秒处理数万消息    HDFS MR数分钟、数小时
  8. 不会主动停止        终有完成的时候。

storm优点

  1. 1.跨语言
  2. 2.可伸缩的
  3. 3.低延迟,秒级/分钟级
  4. 4.容错。

核心概念

  1. 1.Tuple
  2.     主要的数据结构，有序元素的列表。
  3. 2.Stream
  4.     Tuple的序列。
  5. 3.Spouts
  6.     数据流源头。可以读取kafka队列消息。可以自定义。
  7. 4.Bolts
  8.     转接头.
  9.     逻辑处理单元。spout的数据传递个bolt，bolt计算，完成后产生新的数据。
  10.     IBolt是接口。

Topology

  1. Spout + bolt连接在一起形成一个top，形成有向图，定点就是计算，边是数据流。

task

  1. Bolt中每个Spout或者bolt都是一个task.

Storm架构

  1. 1.Nimbus(灵气)
  2.     master节点。
  3.     核心组件，运行top。
  4.     分析top并收集运行task。分发task给supervisor.
  5.     监控top。
  6.     无状态，依靠zk监控top的运行状况。
  8. 2.Supervisor(监察)
  9.     每个supervisor有n个worker进程，负责代理task给worker。
  10.     worker在孵化执行线程最终运行task。
  11.     storm使用内部消息系统在nimbus和supervisor之间进行通信。
  13.     接受nimbus指令，管理worker进程完成task派发。
  15. 3.worker
  16.     执行特定的task，worker本身不执行任务，而是孵化executors，
  17.     让executors执行task。
  19. 4.Executor
  20.     本质上有worker进程孵化出来的一个线程而已。
  21.     executor运行task都属于同一spout或者bolt.
  23. 5.task
  24.     执行实际上的任务处理。或者是Spout或者是bolt.

storm工作流程

  1. 1.nimbus等待提交的top
  2. 2.提交top后，nimbus收集task，
  3. 3.nimbus分发task给所有可用的supervisor
  4. 4.supervisor周期性发送心跳给nimbus表示自己还活着。
  5. 5.如果supervisor挂掉，不会发送心跳给nimubs，nimbus将task发送给其他的supervisor
  6. 6.nimubs挂掉，super会继续执行自己task。
  7. 7.task完成后，supervisor等待新的task
  8. 8.同时，挂掉的nimbus可以通过监控工具软件自动重启。

安装storm集群

  1. [s201 ~ s204]
  2. 1.jdk
  3. 2.tar
  4. 3.环境变量
  5. 4.验证安装
  6.     $>source /etc/profile
  7.     $>./storm version
  8. 5.分发安装文件到其他节点。 
  10. 6.配置
  11.     [storm/conf/storm.yaml]
  12.     ## Nimbus 和 Supervisor 后台进程都需要一个用于存放一些状态数据（比如 jar 包、配置文件等等）的目录
  13.     storm.local.dir: "/home/centos/storm"
  15.     ## Storm 关联的 ZooKeeper 集群的地址列表
  16.     storm.zookeeper.servers:
  17.         - "s202"
  18.         - "s203"
  20.     storm.zookeeper.port: 2181
  22.     ### nimbus.* configs are for the master
  23.     ##用于配置主控节点的地址，可以配置多个。从Storm1.0开始，支持Nimbus的HA。
  24.     nimbus.seeds : ["s201"]
  26.     ### ui.* configs are for the master
  27.     ui.host: 0.0.0.0
  28.     ui.port: 8080
  30.     ##配置每个 Supervisor 机器能够运行的工作进程（worker）数。
  31.     每个 worker 都需要一个单独的端口来接收消息，
  32.     ##这个配置项就定义了 worker 可以使用的端口列表。
  33.     如果你在这里定义了 5 个端口，那么 Storm 就会在该机器上分配最多 5 个worker。
  34.     ##如果定义 3 个端口，那 Storm 至多只会运行三个 worker。
  35.     supervisor.slots.ports:
  36.         - 6700
  37.         - 6701
  38.         - 6702
  39.         - 6703
  40. 7.分发
  42. 8.启动进程
  43.     a)启动s201 nimbus进程
  44.         $>storm nimbus &
  46.     b)启动s202 ~ s204 supervisor进程
  47.         $>storm supervisor &
  49.     c)启动s201的ui进程
  50.         $>storm ui &
  52. 9.通过webui查看
  53.     http://s201:8080/

编程实现CallLog日志统计

  1. 0.pom.xml
  2.     <?xml version="1.0" encoding="UTF-8"?>
  3.     <project xmlns="http://maven.apache.org/POM/4.0.0"
  4.              xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
  5.              xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
  6.         <modelVersion>4.0.0</modelVersion>
  8.         <groupId>com.it18zhang</groupId>
  9.         <artifactId>StormDemo</artifactId>
  10.         <version>1.0-SNAPSHOT</version>
  12.         <dependencies>
  13.             <dependency>
  14.                 <groupId>org.apache.storm</groupId>
  15.                 <artifactId>storm-core</artifactId>
  16.                 <version>1.0.3</version>
  17.             </dependency>
  18.         </dependencies>
  20.     </project>
  21. 1.创建Spout
  22.     package com.it18zhang.stormdemo;
  24.     import org.apache.storm.spout.SpoutOutputCollector;
  25.     import org.apache.storm.task.TopologyContext;
  26.     import org.apache.storm.topology.IRichSpout;
  27.     import org.apache.storm.topology.OutputFieldsDeclarer;
  28.     import org.apache.storm.tuple.Fields;
  29.     import org.apache.storm.tuple.Values;
  31.     import java.util.ArrayList;
  32.     import java.util.List;
  33.     import java.util.Map;
  34.     import java.util.Random;
  36.     /**
  37.      * Spout类,负责产生数据流
  38.      */
  39.     public class CallLogSpout implements IRichSpout{
  41.         //Spout输出收集器
  42.         private SpoutOutputCollector collector;
  44.         //是否完成
  45.         private boolean completed = false;
  47.         //上下文
  48.         private TopologyContext context;
  50.         //随机发生器
  51.         private Random randomGenerator = new Random();
  53.         //
  54.         private Integer idx = 0;
  56.         public void open(Map conf, TopologyContext context, SpoutOutputCollector collector) {
  57.             this.context = context;
  58.             this.collector = collector;
  59.         }
  61.         public void close() {
  62.         }
  64.         public void activate() {
  65.         }
  67.         public void deactivate() {
  69.         }
  71.         /**
  72.          * 下一个元组
  73.          */
  74.         public void nextTuple() {
  75.             if (this.idx <= 1000) {
  76.                 List<String> mobileNumbers = new ArrayList<String>();
  77.                 mobileNumbers.add("1234123401");
  78.                 mobileNumbers.add("1234123402");
  79.                 mobileNumbers.add("1234123403");
  80.                 mobileNumbers.add("1234123404");
  82.                 Integer localIdx = 0;
  83.                 while (localIdx++ < 100 && this.idx++ < 1000) {
  84.                     //取出主叫
  85.                     String caller = mobileNumbers.get(randomGenerator.nextInt(4));
  86.                     //取出被叫
  87.                     String callee = mobileNumbers.get(randomGenerator.nextInt(4));
  88.                     while (caller == callee) {
  89.                         //重新取出被叫
  90.                         callee = mobileNumbers.get(randomGenerator.nextInt(4));
  91.                     }
  92.                     //模拟通话时长
  93.                     Integer duration = randomGenerator.nextInt(60);
  95.                     //输出元组
  96.                     this.collector.emit(new Values(caller, callee, duration));
  97.                 }
  98.             }
  99.         }
  101.         public void ack(Object msgId) {
  103.         }
  105.         public void fail(Object msgId) {
  107.         }
  109.         /**
  110.          * 定义输出的字段名称
  111.          */
  112.         public void declareOutputFields(OutputFieldsDeclarer declarer) {
  113.             declarer.declare(new Fields("from", "to", "duration"));
  114.         }
  116.         public Map<String, Object> getComponentConfiguration() {
  117.             return null;
  118.         }
  119.     }
  121. 2.创建CreatorBolt
  122.     package com.it18zhang.stormdemo;
  124.     import org.apache.storm.task.OutputCollector;
  125.     import org.apache.storm.task.TopologyContext;
  126.     import org.apache.storm.topology.IRichBolt;
  127.     import org.apache.storm.topology.OutputFieldsDeclarer;
  128.     import org.apache.storm.tuple.Fields;
  129.     import org.apache.storm.tuple.Tuple;
  130.     import org.apache.storm.tuple.Values;
  132.     import java.util.Map;
  134.     /**
  135.      * 创建CallLog日志的Bolt
  136.      */
  137.     public class CallLogCreatorBolt implements IRichBolt {
  138.         //
  139.         private OutputCollector collector;
  141.         public void prepare(Map stormConf, TopologyContext context, OutputCollector collector) {
  142.             this.collector = collector ;
  143.         }
  145.         public void execute(Tuple tuple) {
  146.             //处理通话记录
  147.             String from = tuple.getString(0);
  148.             String to = tuple.getString(1);
  149.             Integer duration = tuple.getInteger(2);
  150.             //产生新的tuple
  151.             collector.emit(new Values(from + " - " + to, duration));
  152.         }
  154.         public void cleanup() {
  156.         }
  158.         /**
  159.          * 设置输出字段的名称
  160.          */
  161.         public void declareOutputFields(OutputFieldsDeclarer declarer) {
  162.             declarer.declare(new Fields("call", "duration"));
  163.         }
  165.         public Map<String, Object> getComponentConfiguration() {
  166.             return null;
  167.         }
  168.     }
  170. 3.创建CounterBolt
  171.     package com.it18zhang.stormdemo;
  173.     import org.apache.storm.task.IBolt;
  174.     import org.apache.storm.task.OutputCollector;
  175.     import org.apache.storm.task.TopologyContext;
  176.     import org.apache.storm.topology.IRichBolt;
  177.     import org.apache.storm.topology.OutputFieldsDeclarer;
  178.     import org.apache.storm.tuple.Fields;
  179.     import org.apache.storm.tuple.Tuple;
  181.     import java.util.HashMap;
  182.     import java.util.Map;
  184.     /**
  185.      * 通话记录计数器Bolt
  186.      */
  187.     public class CallLogCounterBolt implements IRichBolt{
  189.         Map<String, Integer> counterMap;
  190.         private OutputCollector collector;
  192.         public void prepare(Map stormConf, TopologyContext context, OutputCollector collector) {
  193.             this.counterMap = new HashMap<String, Integer>();
  194.             this.collector = collector;
  195.         }
  197.         public void execute(Tuple tuple) {
  198.             String call = tuple.getString(0);
  199.             Integer duration = tuple.getInteger(1);
  201.             if (!counterMap.containsKey(call)) {
  202.                 counterMap.put(call, 1);
  203.             } else {
  204.                 Integer c = counterMap.get(call) + 1;
  205.                 counterMap.put(call, c);
  206.             }
  207.             collector.ack(tuple);
  208.         }
  210.         public void cleanup() {
  211.             for (Map.Entry<String, Integer> entry : counterMap.entrySet()) {
  212.                 System.out.println(entry.getKey() + " : " + entry.getValue());
  213.             }
  214.         }
  216.         public void declareOutputFields(OutputFieldsDeclarer declarer) {
  217.             declarer.declare(new Fields("call"));
  218.         }
  220.         public Map<String, Object> getComponentConfiguration() {
  221.             return null;
  222.         }
  223.     }
  225. 4.App
  226.     package com.it18zhang.stormdemo;
  228.     import org.apache.storm.Config;
  229.     import org.apache.storm.LocalCluster;
  230.     import org.apache.storm.topology.TopologyBuilder;
  231.     import org.apache.storm.tuple.Fields;
  233.     /**
  234.      * App
  235.      */
  236.     public class App {
  237.         public static void main(String[] args) throws InterruptedException {
  238.             TopologyBuilder builder = new TopologyBuilder();
  239.             //设置Spout
  240.             builder.setSpout("spout", new CallLogSpout());
  241.             //设置creator-Bolt
  242.             builder.setBolt("creator-bolt", new CallLogCreatorBolt()).shuffleGrouping("spout");
  243.             //设置counter-Bolt
  244.             builder.setBolt("counter-bolt", new CallLogCounterBolt()).fieldsGrouping("creator-bolt", new Fields("call"));
  246.             Config conf = new Config();
  247.             conf.setDebug(true);
  249.             LocalCluster cluster = new LocalCluster();
  250.             cluster.submitTopology("LogAnalyserStorm", conf, builder.createTopology());
  251.             Thread.sleep(10000);
  253.             //停止集群
  254.             cluster.shutdown();
  255.         }
  256.     }
  259. 5.在生产环境的集群上部署storm top
  260.     a)修改提交方式
  261.         [App.java]
  262.          public static void main(String[] args) throws Exception {
  263.                 TopologyBuilder builder = new TopologyBuilder();
  264.                 //设置Spout
  265.                 builder.setSpout("spout", new CallLogSpout());
  266.                 //设置creator-Bolt
  267.                 builder.setBolt("creator-bolt", new CallLogCreatorBolt()).shuffleGrouping("spout");
  268.                 //设置counter-Bolt
  269.                 builder.setBolt("counter-bolt", new CallLogCounterBolt()).fieldsGrouping("creator-bolt", new Fields("call"));
  271.                 Config conf = new Config();
  272.                 conf.setDebug(true);
  274.                 /**
  275.                  * 本地模式storm
  276.                  */
  277.         //        LocalCluster cluster = new LocalCluster();
  278.         //        cluster.submitTopology("LogAnalyserStorm", conf, builder.createTopology());
  279.         //        Thread.sleep(10000);
  280.                 StormSubmitter.submitTopology("mytop", conf, builder.createTopology());
  281.             }
  282.     b)导入jar包.
  283.         maven ...
  285.     c)在centos上运行top
  286.         $>storm jar xxx.jar com.it18zhang.stormdemo.App

使用storm流计算实现wordcount

  1. 1.WordCountSpout
  2.     ...
  3. 2.SplitBolt
  4.     String line  = ...
  5.     String[] str = line.split(" ");
  6.     for(String s : str){
  7.         collector.emit(new Values("word","1"))
  8.     }
  9. 3.CounterBolt

设置top的并发程度和任务

  1. 配置并发度.
  2. 1.设置worker数据
  3.     conf.setNumWorkers(1);
  5. 2.设置executors个数
  6.     //设置Spout的并发暗示 (executor个数)
  7.     builder.setSpout("wcspout", new WordCountSpout(),3);
  9.     //设置bolt的并发暗示
  10.     builder.setBolt("split-bolt", new SplitBolt(),4)
  12. 3.设置task个数
  13.     每个线程可以执行多个task.
  14.     builder.setSpout("wcspout", new WordCountSpout(),3).setNumTasks(2);
  15.     //
  16.     builder.setBolt("split-bolt", new SplitBolt(),4).shuffleGrouping("wcspout").setNumTasks(3);
  18. 4.并发度 ==== 所有的task个数的总和。