大数据-Storm

Storm 流式处理框架

Storm是实时的,分布式,高容错的计算系统。java+cljoure

Storm常驻内存,数据在内存中处理不经过磁盘,数据通过网络传输。

底层java+cljoure构成,阿里使用java重构Storm构建Jstorm。

数据处理分类

  • 流式处理(异步)

    • 客户端提交数据进行结算,不会等待计算结果

    • 数据追条处理:数据清洗或分析

    • 例:在数据统计分析中:数据存入队列,storm从MQ获取数据,storm将结果存入存储层

  • 实时请求应答(同步)

    • 客户端提交请求后立即计算并将结果返回给客户端

    • 图片特征提取: 客户端向Storm的drpc发送请求,客户端等待Storm响应结果

Storm与MR的比较

  • Storm流式处理,毫秒级响应,常驻内存

  • MR批处理TB级数据,分钟级响应,反复启停

优势

高可靠:异常处理;消息可靠性保障机制

可维护性:提供UI监控接口

架构模型

  • Nimbus:接收请求,接收jar包并解析,任务分配,资源分配

  • zookeeper:存储Nimbus分配的子任务,存储任务数据,用于解耦架构

  • Supervisor:从zk中获取子任务,启动worker任务,启停所管理的worker

  • Worker:执行具体任务(每个worker对应一个Topology的子集),分为spout任务和bolt任务,启动executor,一个executor对应一个线程,默认一个executor执行一个task任务。任务所需数据从zookeeper中获取。

storm与MR比对 MR Storm
主节点 ResourceManager Nimbus
从节点 NodeManager Supervisor
应用程序 Job Topology
工作进程 Child Worker
计算模型 Map/Reduce Spout/Bolt

编程模型

DAG有向无环图,拓扑结构图,作为实时计算逻辑的封装。一旦启动会一直在系统中运行直到手动kill

大数据-Storm

Spout 数据源节点

  • 从外部读取原始数据

  • 定义数据流,可定义多个数据流向不同Bolt 发数据,读取外部数据源

  • nextTuple:被Storm线程不断调用、主动从数据源拉取数据,再通过emit方法将数据生成Tuple,发送给之后的Bolt计算

Bolt 计算节点

  • execute:该方法接收到一个Tuple数据,真正业务逻辑

  • 定义数据流,将结果数据发出去

API

以wordcount为例:Spout 获取原始行数据,将行数据Tuple作为发给bolt1,bolt1解析行数据为单词数据,bolt1将单词数据转为Tuple发给bolt2,bolt2对单词数据进行计数,将计数结果发送到存储层。

Topology类

指定计算任务的各节点及节点关联,提交任务

  • topologyBuilder的setSpout方法和setBolt方法构建各节点

    • 方法参数为:id值,bolt类,并发度(线程),task数,数据来源节点,数据流,分组字段

    • shuffleGrouping是随机分配数据

    • 多并发度

      • shuffleGrouping是将数据随机分配给线程

      • fieldsGrouping是根据指定的若干字段分组到不同的线程

  • Config对象传入配置参数

    • 超时时间 秒:conf.setMessageTimeoutSecs(3);

  • LocalCluster本地执行,对象提交任务名,参数对象,拓扑对象,最终执行任务

public class MyTopology {
   public static void main(String[] args) {
       //创建拓扑构建对象
       TopologyBuilder topologyBuilder=new TopologyBuilder();
       /**
        * 指定各节点,线程数量,上下游关联
        * 对于多线程,注意上游的结果会随机发给各个线程
        */
       //设置spout节点,指定(id值,spout类,并发度)
       topologyBuilder.setSpout("s1",new CountSpout(),1);
       //设置bolt节点,指定 (id值,bolt类,并发度,数据来源节点,分组字段)
       topologyBuilder.setBolt("b1",new CountBolt1(),2).setNumTasks(4).shuffleGrouping("s1");
       topologyBuilder.setBolt("b2",new CountBolt2(),1).fieldsGrouping("b1",new Fields("word"));
       //构建拓扑对象
       StormTopology stormTopology=topologyBuilder.createTopology();
       //设置配置信息,该对象实际是个map,向其中设置各种参数
       Config config=new Config();
       config.setNumWorkers(3);
       config.put("xxx","xxx");
       //获取本地执行集群对象
       LocalCluster lc=new LocalCluster();
       //本地执行:向集群提交任务,任务名,参数,拓扑节点
       lc.submitTopology("job-count",config,stormTopology);
  }
}

Spout类(自定义)

继承BaseRichSpout类,实际实现了IRichSpout接口,主要重写以下方法

  • open()——初始化,参数为:配置参数, 上下文,输出对象

  • nextTuple()——业务代码

  • declareOutputFields()——定义输出结果

  • 关于数据输出

    将SpoutOutputCollector输出对象转为全局变量,通过SpoutOutputCollector输出数据

    使用Values对象,存入多个结果(相当于一个list)

    通过OutputFieldsDeclarer对象,依次定义每个结果的字段名(相当于给list字段命名,形成类似map的有序集合)

import backtype.storm.topology.base.BaseRichSpout;
import backtype.storm.tuple.Fields;
import backtype.storm.tuple.Values;
public class CountSpout extends BaseRichSpout {
   /** 将输出对象转为全局属性,支持数据输出 */
   private SpoutOutputCollector collector;
   //初始化
   @Override
   public void open(Map conf, TopologyContext context, SpoutOutputCollector collector) {
       //将输出对象转为全局属性
       this.collector=collector;
  }
   //业务代码:随机输出一行字符串
   @Override
   public void nextTuple() {
       int index=random.nextInt(lines.length);
       String line= lines[index];
       
       //输出数据,构建value时能够设置多个数据,在Fields对象中定义字段名
       collector.emit(new Values(line,line));
  }
   @Override
   public void declareOutputFields(OutputFieldsDeclarer declarer) {
       //定义输出数据,给每个字段赋予字段名
       declarer.declare(new Fields("line1","line2"));
  }
}

Bolt类(自定义)

继承BaseRichBolt类,实际实现了IRichBolt接口,主要重写以下方法

  • prepare()——初始化

  • execute()——业务代码

  • declareOutputFields()——定义输出字段

  • 关于输出数据,与spout相同

  • 关于输入数据:通过input的各种方法获取对应类型的数据,支持通过字段所在索引下标或字段名获取指定位置的值

public class CountBolt1 extends BaseRichBolt {
/** 将输出对象转为全局属性,支持数据输出 */
   private OutputCollector collector;
   @Override
   public void prepare(Map stormConf, TopologyContext context, OutputCollector collector) {
       this.collector=collector;
  }
   @Override
   public void execute(Tuple input) {
       /* 不同方法获取不同类型的值,方法中通过索引下标或字段名来取值 */
       //通过索引取值
       String line1=input.getString(0);
       //通过字段名取值
       String line2=input.getStringByField("line1");

       //以下为业务逻辑,将字符串拆成单词数组,遍历输出
       String[] words=line1.split(" ");
       for (String word:words){
           //遍历输出,输出两个字段的数据
           collector.emit(new Values(word,word));
      }
  }
   @Override
   public void declareOutputFields(OutputFieldsDeclarer declarer) {
       declarer.declare(new Fields("word","word2"));
  }
}

分组策略

  • Shuffle Grouping:轮询的方式,随机派发tuple,每个线程接收到数量大致相同。

  • Fields Grouping:按指定的字段分组,指定字段的值相同tuple,分配相同的task(处于同一个线程中)

  • 以下方式了解

    • All Grouping:广播发送,所有下游的bolt线程都会收到。还是task

    • Global Grouping:全局分组, 将tuple分配给 id最低的task,这个id是用户指定的id值。

    • None Grouping:随机分组

    • Direct Grouping:指向型分组

    • local or shuffle grouping:若目标bolt task与源bolt task在同一工作进程中,则tuple随机分配给同进程中的tasks。否则普通的Shuffle分配 Grouping行为一致

    • customGrouping:自定义

分布式搭建

环境搭建

  • 安装

    tar -zxvf apache-storm-0.10.0.tar.gz

    mv apache-storm-0.10.0 storm-0.10.0

  • 修改配置

    vim /opt/sxt/storm-0.10.0/conf/storm.yaml

    #指定zk地址
    storm.zookeeper.servers:
        - "node1"
        - "node2"
        - "node3"
       
    #指定nimbus的节点
    nimbus.host: "node1"

    #指定开放的端口(默认以下4个)
    supervisor.slots.ports:
      - 6700
      - 6701
      - 6702
      - 6703
  • 拷贝到其他节点

    cd /opt/sxt 
    scp -r storm-0.10.0/ node3:`pwd`

启动+命令

启动集群

  • 启动zookeeper集群

    zkServer.sh start

  • 启动nimbus节点 cd /opt/sxt/storm-0.10.0

    • 各节点创建logs目录:mkdir logs

    • 主节点启动:

      • 启动nimbus

        nohup ./bin/storm nimbus >> logs/nimbus.out 2>&1 &

      • 启动监控ui

        nohup ./bin/storm ui >> logs/ui.out 2>&1 &

    • 从节点启动supervisor

    • nohup ./bin/storm supervisor >> ./logs/supervisor.out 2>&1 &

    • 注:项目的日志也就存放中logs目录下的各文件中

      可以远程查看:例如文件wc-1-1571317142-worker-6700.log

      http://node1:8080/topology.html?wc-1-1571317142-worker-6700.log

  • 网页登录设置在主节点上的UI监控地址 http://node1:8080

关闭集群

#关闭nimbus
kill `ps aux | egrep '(daemon.nimbus)|(storm.ui.core)' | fgrep -v egrep | awk '{print $2}'`
#关闭supervisor
kill `ps aux | fgrep storm | fgrep -v 'fgrep' | awk '{print $2}'`
#关闭zk
zkServer.sh stop

运行项目

从IDE中导出项目jar文件(field标签 > project structure > 指定主类创建jar配置 > bulid标签 > build project > 从项目目录中找到jar文件) 导入到/opt/sxt下

载入项目命令:./bin/storm jar /opt/sxt/wc.jar com.shsxt.storm.wordcount.WordCountTP wc

依次指定jar文件目录,主类,topology名

停止项目命令:./bin/storm kill wc (注意指定topology)

内部文件结构

  • 本地目录

    /opt/sxt/storm-0.10.0/storm-local 目录下存放项目执行时的数据

    内部为一个目录树结构

  • zookeeper目录

    进入zk客户端 zkCli.sh

    查看storm的目录 ls /storm 内部时一个目录树

    查看指定的数据 get /xxx/xxx

并发机制

  • worker进程

    • 具体执行任务的进程

    • 一个topology包含若干个worker执行,worker执行topology的子集

    • 一个worker需要绑定一个网络端口,默认一个节点最多运行4个worker,通过配置多于4个端口开放更多worker

  • Executor线程

    • execute是由worker进程创建的线程,worker包含至少一个executor线程

    • 一个线程中能够执行一个或多个task任务,默认一个线程执行一个task

    • spout / boltk 与线程是一对多的关系,一个spout 可以包含多个线程,但线程只对应一个spout,监控线程除外。

  • task

    • 数据处理的最小单元

    • 每个task对应spout或bolt的一个并发度

关于再平衡热rebalance

  • 在topology生命周期中,worker与executor的数量可变,task的数量不可变

  • 多个线程无法同时执行一个task,只能一个线程执行一个task,因此线程数应该小于task数,设置多的线程是无效的。

  • 注意:系统对每个worker都会设置一个监控线程,监控线程数量与worker相同

  • 命令修改

    ./bin/storm/rebalance wc -n 5 -e blue-spout=3 -e yellow-bolt=10

    (调整wc任务 5个worker 3个spout线程 10个bolt线程 )

  • api设置

    • worker数

      通过配置文件制定worker数量

      config.setNumWorkers(3);

    • executor数与task数量

      线程数设置 setBolt(String id, IRichBolt bolt, Number 线程数);和setSpout()方法

      设置任务数 setNumTasks(task数)

      topologyBuilder.setBolt("wcSplit",new WordCountSplit(),2).setNumTasks(4).shuffleGrouping("wcSpout");

数据传输

  • worker之间数据传输使用Netty架构

    • 接收线程监听tcp端口,并将数据存入缓冲区域

    • 发送线程使用传输队列发送数据

  • worker的内部通讯采用Disruptor的队列处理机制

    数据从缓冲区读取经过:接收队列——工作线程——发送队列——发送线程

  • tuple 在Storm中的数据的传输单元

  • Stream 数据连通的管道,多个管道时需要指定Stream的id

线程数包括:spout线程数,blot线程数,acker(每个work1个监控线程)

容错机制

宕机机制

  • nimbus宕机

    • 无状态策略,所有的状态信息都存放在Zookeeper中来管理,worker正常运行,只是无法提交新的topology

    • 快速失败策略:自检到异常后自毁并重启

  • supervisor宕机

    • 无状态策略,supervisor的所有的状态信息都存放在Zookeeper中来管理

    • 快速失败策略:自检到异常后自毁并重启

    • 节点故障状态下,若所有该节点上的任务超时,nimbus会将该节点的task分配给其他的节点

  • worker宕机

    • supervisor会重启worker进程

    • 若worker多次失败且无法向nimbus发送心跳,nimbus会将该worker的分配到其他supervisor上执行。

数据跟踪acker

消息完整性

spout中发出的tuple及其由tuple产生的子tuple,构成一个树形结构,只有当数中的所有消息都被正确处理,才能保证数据的完整性。

为了保证数据的完整性,对于处理失败的tuple需要请求上游的spout重发数据。

消息重发的机制:at least(至少一次),oncely (有且只有一次)

acker机制

acker跟踪每个spout发出的tuple。

acker记录每个trulp的id,数据处理后,通过亦或算法清空id,若id未清空则表明这条数据没有处理完成。

  • 自定义acker类,对数据进行监控

    • 在topology中添加具备acker功能的spout作为acker类

      topologyBuilder.setSpout("myack",new MyAckSpout());

    • 定义acker类

      • 实现IRichSpout接口,重写ack(成功)和fail(失败)方法,对发出的数据进程缓存

      • 设置全局缓存,数据发送时存入缓存,处理成功删除缓存,处理失败,从缓存取数据重发

      public class MyAckSpout implements IRichSpout {
         SpoutOutputCollector collector;
         //数据缓存
         Map<Object,String> map =  new HashMap<>();
         //设置全局缓存id
         long id = 0;
         @Override
         public void nextTuple() {
             //发送数据,并存入消息id
             collector.emit(new Values("你好"),id);
             //数据存入缓存
             map.put(id,line);
             //索引自增处理
             id++; }

         //数据完整处理调用ack方法,参数为消息id索引
         @Override
         public void ack(Object msgId) {
             //将缓存中的数据清除
             map.remove(msgId); }

         //数据没有完整处理调用fail方法,参数为消息id索引
         @Override
         public void fail(Object msgId) {
             //重发消息
             collector.emit(new Values(map.get(msgId)),msgId);
        }
      //省略以下重写方法open,declareOutputFields,getComponentConfiguration,close,activate,deactivate
      }
  • bolt 中代码响应,execute方法中

    • 方法末尾添加数据处理的响应

      处理成功:collection.ack(input)

      处理失败:collection.fail(input) 注意:超时也数据失败

    • 向下游发送数据时,指定acker是否继续跟踪下游数据

      跟踪:collector.emit(input,new Values(split[i]));

      不跟踪:collector.emit(new Values(split[i]));

      public void execute(Tuple input) {
         String xxx = input.getStringByField("aaa");
         //结束跟踪
         collector.emit(new Values(xxx));
         //继续collector.emit(new Values(xxx));
      //处理成功
         collector.ack(input);
         //处理失败,collector.fail(input);
      }

实时计算整合

采集层(flume):实现日志收集,使用负载均衡策略

消息队列(kafka):作用是解耦及不同速度系统缓冲

实时处理单元(Storm):用Storm来进行数据处理,最终数据流入DB中

展示单元(DB+web):数据可视化,使用WEB框架展示

整合kafka

 

 

 

 

 

上一篇:Linux删除文件内容的方法


下一篇:Storm本地启动拓扑报错:Exception in thread "main" java.lang.NoClassDefFoundError: org/apache/storm/