大数据系列——Storm安装和API

时间 2019-12-07

标签数据系列 storm 安装 api 栏目 Storm 繁體版

原文原文链接

1. 实时计算

有别于传统的离线批处理操做(对不少数据的集合进行的操做)
实时处理，说白就是针对一条一条的数据/记录进行操做
实时计算计算的是无界数据

2. 有界数据和无界数据

2.1 有界数据

离线计算面临的操做数据都是有界限的，不管是1G、1T、1P、1EB、1NB
数据的有界必然会致使计算的有界

2.2 无界数据

实时计算面临的操做数据是源源不断的向水流同样，是没有界限的
数据的无界必然致使计算的无界

3. 计算中心和计算引擎

在大数据领域中存在三大计算中心和三大计算引擎java

3.1 三大计算中心

离线计算计算中心(mapreduce)
实时计算中心(storm flink...)
准实时计算中心(spark)

3.2 三大计算引擎

交互式查询计算引擎(hive sparksql)
图计算计算引擎
机器学习计算引擎

4. Storm简介

免费开源分布式实时计算系统
处理无界的数据流
Tiwtter开源的cloujre
Storm能实现高频数据和大规模数据的实时处理
官网资料显示storm的一个节点1秒钟可以处理100万个100字节的消息(IntelE5645@2.4Ghz的CPU,24GB的内存)
storm是毫秒级的实时处理框架

Apache Storm是Twitter开源的一个相似于Hadoop的实时数据处理框架，它原来是由BackType开发，后BackType被Twitter收购，将Storm做为Twitter的实时数据分析系统。sql

5. hadoop与storm的计算

数据来源shell
- hadoop数据库
  - HADOOP处理的是HDFS上TB级别的数据(历史数据)
- stormapache
  - STORM是处理的是实时新增的某一笔数据(实时数据)

处理过程安全
- hadoopbash
  - HADOOP是分MAP阶段到REDUCE阶段
  - HADOOP最后是要结束的
- storm架构
  - STORM是由用户定义处理流程，流程中能够包含多个步骤，每一个步骤能够是数据源(SPOUT)或处理逻辑(BOLT)
  - STORM是没有结束状态，到最后一步时，就停在那，直到有新数据进入时再从头开始

处理速度框架
- hadoopssh
  - HADOOP是以处理HDFS上TB级别数据为目的，处理速度慢
- storm
  - STORM是只要处理新增的某一笔数据便可，能够作到很快 (毫秒级的响应)

适用场景
- HADOOP是在要处理批量数据时用的，不讲究时效性
- STORM是要处理某一新增数据时用的，要讲时效性

6. Storm的架构

Spout
- Storm认为每一个stream都有一个stream源，也就是原始元组的源头，因此它将这个源头称为Spout
- 消息源，是消息生产者，他会从一个外部源读取数据并向topology里面面发出消息
Bolt
- 消息处理者，全部的消息处理逻辑被封装在bolts里面，处理输入的数据流并产生新的输出数据流,可执行过滤，聚合，查询数据库等操做
数据流
Task 每个Spout和Bolt会被看成不少task在整个集群里面执行,每个task对应到一个线程.
Stream groupings: 消息分发策略,定义一个Topology的其中一步是定义每一个tuple接受什么样的流做为输入,stream grouping就是用来定义一个stream应该如何分配给Bolts们.

7. Storm集群的安装

准备安装文件
apache-storm-1.0.2.tar.gz
解压

[root@uplooking01 /soft]
    tar -zxvf apache-storm-1.0.2.tar.gz -C /opt
    mv apache-storm-1.0.2/ storm

配置storm

storm-env.sh

[root@uplooking01 /soft]
    export JAVA_HOME=/opt/jdk
    export STORM_CONF_DIR="/opt/storm/conf"

storm.yaml

[root@uplooking01 /opt/storm/conf]


storm.zookeeper.servers:
  - "uplooking03"
  - "uplooking04"
  - "uplooking05"

#配置两个主节点,实现主节点的单点故障
nimbus.seeds: ["uplooking01", "uplooking02"]
storm.local.dir: "/opt/storm/storm-local"
#配置从节点的槽数
supervisor.slots.ports:
  - 6700
  - 6701
  - 6702
  - 6703

分发到其余节点

[root@uplooking01 /]
    scp -r /opt/storm uplooking02:/opt
    scp -r /opt/storm uplooking03:/opt
    scp -r /opt/storm uplooking04:/opt
    scp -r /opt/storm uplooking05:/opt

启动storm

[root@uplooking01 /]  
    #启动主进程和ui进程
    nohup /opt/storm/bin/storm nimbus >/dev/null 2>&1 &
    nohup /opt/storm/bin/storm ui >/dev/null 2>&1 &
    nohup /opt/storm/bin/storm logviewer >/dev/null 2>&1 &

[root@uplooking02 /]
    #启动主进程(numbus)
    nohup /opt/storm/bin/storm numbus >/dev/null 2>&1 &
    nohup /opt/storm/bin/storm logviewer >/dev/null 2>&1 &

#启动从节点进程(supervisor)
[root@uplooking03 /]
    nohup /opt/storm/bin/storm supervisor >/dev/null 2>&1 &
    nohup /opt/storm/bin/storm logviewer >/dev/null 2>&1 &
[root@uplooking04 /]
    nohup /opt/storm/bin/storm supervisor >/dev/null 2>&1 &
    nohup /opt/storm/bin/storm logviewer >/dev/null 2>&1 &
[root@uplooking05 /]
    nohup /opt/storm/bin/storm supervisor >/dev/null 2>&1 &
    nohup /opt/storm/bin/storm logviewer >/dev/null 2>&1 &

8. Storm集群的启动脚本

#!/bin/bash
#启动nimbus

for nimbusHost in  `cat /opt/shell/nimbus.host`
do
#-T 进制分配伪终端 通常自动化脚本不须要分配伪终端
ssh -T  root@${nimbusHost}    << eeooff
    nohup /opt/storm/bin/storm nimbus >/dev/null 2>&1 &
eeooff
done

#启动supervisor
for supervisorHost in  `cat /opt/shell/supervisor.host`
do
#-T 进制分配伪终端 通常自动化脚本不须要分配伪终端
ssh -T  root@${supervisorHost}    << eeooff
        nohup /opt/storm/bin/storm supervisor >/dev/null 2>&1 &
eeooff
done


#启动logviewer
for logviewerHost in  `cat /opt/shell/logviewer.host`
do
#-T 进制分配伪终端 通常自动化脚本不须要分配伪终端
ssh -T  root@${logviewerHost}    << eeooff
        nohup /opt/storm/bin/storm logviewer >/dev/null 2>&1 &
eeooff
done


#启动ui
for uiHost in  `cat /opt/shell/ui.host`
do
#-T 进制分配伪终端 通常自动化脚本不须要分配伪终端
ssh -T  root@${uiHost}    << eeooff
        nohup /opt/storm/bin/storm ui >/dev/null 2>&1 &
eeooff
done

9. Storm实现数字累加

编写Spout

public class MySpout extends BaseRichSpout {
    private SpoutOutputCollector collector;
    //初始化累加的数字
    int num = 0;

    @Override
    public void open(Map conf, TopologyContext context, SpoutOutputCollector collector) {
        this.collector = collector;
    }

    @Override
    public void nextTuple() {
        collector.emit(new Values(num));
        num++;
    }

    @Override
    public void declareOutputFields(OutputFieldsDeclarer declarer) {
        declarer.declare(new Fields("mynum"));
    }
}

编写Bolt

public class MyBolt extends BaseRichBolt {
    @Override
    public void prepare(Map stormConf, TopologyContext context, OutputCollector collector) {

    }

    @Override
    public void execute(Tuple tuple) {
        Integer num = tuple.getIntegerByField("mynum");
        System.out.println(num);
    }

    @Override
    public void declareOutputFields(OutputFieldsDeclarer declarer) {

    }
}

编写Topology

public class MyTopology {
    public static void main(String[] args) {
        //建立自定义的spout
        MySpout mySpout = new MySpout();
        //建立自定义的bolt
        MyBolt myBolt = new MyBolt();
        //建立topology名称
        String topologyName = "MyNumTopology";
        //建立topology的配置对象
        Map conf = new Config();

        //建立topology的构造器
        TopologyBuilder topologyBuilder = new TopologyBuilder();
        //为topology设置spout和bolt
        topologyBuilder.setSpout("myspout", mySpout);
        topologyBuilder.setBolt("mybolt", myBolt).shuffleGrouping("myspout");

        //建立本地的topology提交器
        StormTopology stormTopology = topologyBuilder.createTopology();
        LocalCluster localCluster = new LocalCluster();
        localCluster.submitTopology(topologyName, conf, stormTopology);
    }
}

10. 多个Bolt的问题

定义下一个Bolt

public class MyBolt02 extends BaseRichBolt {
    @Override
    public void prepare(Map stormConf, TopologyContext context, OutputCollector collector) {

    }

    @Override
    public void execute(Tuple tuple) {
        System.out.println(tuple.getIntegerByField("mynum02") + ".....");
    }

    @Override
    public void declareOutputFields(OutputFieldsDeclarer declarer) {

    }
}

第一个Bolt中给第二个Bolt发射数据

public class MyBolt extends BaseRichBolt {

    private OutputCollector collector;

    @Override
    public void prepare(Map stormConf, TopologyContext context, OutputCollector collector) {
        this.collector = collector;
    }

    @Override
    public void execute(Tuple tuple) {
        Integer num = tuple.getIntegerByField("mynum");
        System.out.println(num);
        collector.emit(new Values(num));
    }

    @Override
    public void declareOutputFields(OutputFieldsDeclarer declarer) {
        declarer.declare(new Fields("mynum02"));
    }
}

在Topology中配置第二个Bolt

public class MyTopology {
    public static void main(String[] args) {
        //建立自定义的spout
        MySpout mySpout = new MySpout();
        //建立自定义的bolt
        MyBolt myBolt = new MyBolt();

        MyBolt02 myBolt02 = new MyBolt02();
        //建立topology名称
        String topologyName = "MyNumTopology";
        //建立topology的配置对象
        Map conf = new Config();

        //建立topology的构造器
        TopologyBuilder topologyBuilder = new TopologyBuilder();
        //为topology设置spout和bolt
        topologyBuilder.setSpout("myspout", mySpout);
        topologyBuilder.setBolt("mybolt", myBolt).shuffleGrouping("myspout");
        topologyBuilder.setBolt("mybolt02", myBolt02).shuffleGrouping("mybolt");

        //建立本地的topology提交器
        StormTopology stormTopology = topologyBuilder.createTopology();
        LocalCluster localCluster = new LocalCluster();
        localCluster.submitTopology(topologyName, conf, stormTopology);
    }
}

11. 提交做业到集群

StormSubmitter.submitTopology(topologyName, conf, stormTopology);

12. Storm的并行度

在storm中的并行度说的就是一个进程的运行须要多少个线程来参与，若是storm运行的线程个数+1，则并行度+1

Worker :

worker是一个进程级别的概念，能够经过jps查看的到
worker是一个Topology实例的子集，也就是说一个Topology的实例在supervisor中运行，能够在一个或者多个supervisor中启动一个或者多个worker进程
一个worker进程只能为一个Topology实例服务
因此Topology和worker的关系===>1:N
进程是由多个线程来组成，这里的线程就是Executor
conf.setNumWorkers(int workers)
因此worker和executor的关系===>1:N
每个executor线程具体干活是由一个个task任务的实例来完成的
在builer.setSpout/setBolt的第三个参数设置
Task真正在topology干活的实例，一个executor线程，默认状况下对应了1个task的实例的
Executor和Task的关系===>1:N
builder.setSpout().setNumTasks(tasks)//设置的是spout对应的executor拥有几个task实例builder.setBolt().setNumTasks(tasks)//设置的是bolt对应的executor拥有几个task实例

13. Storm中的消息确认机制

在spout中若是发送消息时指定messageId则表明开启消息确认机制,若是不指定messageID则表明不开启消息确认机制
若是Spout中开启了消息确认机制则在bolt中须要用ack()方法来确认消息接收成功
在Soput中重写响应的fail()和ack()方法来处理消息成功或者失败的回调逻辑
Storm默认若是不确认消息接收成功则30s以后返回消息失败
消息确认机制要慎重使用(效率换取安全)