Storm工作原理&集群环境搭建

阅读量：2068 次

发布时间：2019-04-29

本文共 6740 字，大约阅读时间需要 22 分钟。

序言

Storm是Twitter开源的分布式实时大数据处理框架，最早开源于github，从0.9.1版本之后，归于Apache社区，被业界称为实时版Hadoop。

Apache Storm是一个分布式实时大数据处理系统。Storm设计用于在容错和水平可扩展方法中处理大量数据。

参考资料:

应用场景

随着越来越多的场景对Hadoop的MapReduce高延迟无法容忍，比如网站统计、推荐系统、预警系统、金融系统(高频交易、股票)等等，大数据实时处理解决方案（流计算）的应用日趋广泛，目前已是分布式技术领域最新爆发点，而Storm更是流计算技术中的佼佼者和主流。

特点

适用场景广泛：实时处理消息以及更新数据库，基于最基础的实时计算语义和API（实时数据处理领域）；对实时的数据流持续的进行查询或计算，同时将最新的计算结果持续的推送给客户端展示，同样基于最基础的实时计算语义和API（实时数据分析领域）；对耗时的查询进行并行化，基于DRPC，即分布式RPC调用，单表30天数据，并行化，每个进程查询一天数据，最后组装结果(持续接收数据,持续计算,持续产出,这个么意思说了一大堆.cuiyaonan2000@163.com)

可伸缩性高: 如果要扩容，直接加机器，调整storm计算作业的并行度就可以了，storm会自动部署更多的进程和线程到其他的机器上去，无缝快速扩容。(原因:Storm使用ZooKeeper来协调集群内的各种配置使得Storm的集群可以很容易的扩展。)

保证无数据丢失：实时系统必须保证所有的数据被成功的处理。那些会丢失数据的系统的适用场景非常窄，而storm保证每一条消息都会被处理，------目前不知道怎么做到的,谁知道cuiyaonan2000@163.com

异常健壮： storm集群非常容易管理，轮流重启节点不影响应用。因为元数据全部放zookeeper，不在内存中，随便挂都不要紧(这是因为任务信息都是元数据存储在Zookeeper上,任务执行失败可以换到另一台机器上)

容错性好：在消息处理过程中出现异常， storm会进行重试

语言无关性： Storm的topology和消息处理组件(Bolt)可以用任何语言来定义，这一点使得任何人都可以使用storm.

内部组件

Nimbus：负责资源分配和任务调度。一个Storm集群只有一个Nimbus。

Supervisor：即Storm的Slave，负责接收Nimbus分配的任务，管理所有Worker，一个Supervisor节点中包含多个Worker进程。

Worker：工作进程，每个工作进程中都有多个Task ------即进程中有多个线程,即Executor

Executor: 工作线程

Task：任务(即java中的线程)，在 Storm 集群中每个 Spout 和 Bolt 都由若干个任务（tasks）来执行。每个任务都与一个执行线程相对应。

Topology：计算拓扑，是一个虚拟概念,代表的是一个计算流程内的所有内容.，拓扑会一直在集群中运行，直到你手动去终止它。拓扑就是由多个 Spout 和 Bolt 组成的总代名词

Stream：数据流,是一个不间断的无界的连续tuple，注意Storm在建模事件流时，把流中的事件抽象为tuple即元组

Tuple: Storm 的核心数据结构是 tuple。 tuple是包含了一个或者多个键值对的列表，Stream 是由无限制的 tuple 组成的序列。

Spout：数据源（Spout）是拓扑中数据流的来源。

Bolt：消息处理逻辑被封装在bolts里面，Bolts可以做很多事情：过滤，聚合，查询数据库等。

Stream grouping：数据流分组定义了在 Bolt 的不同任务（tasks）中划分数据流的方式。在 Storm 中有八种内置的数据流分组方式。(Stream 分类,并且有8种内置的,还可以创建自定义的)

Reliability：可靠性。Storm 可以通过拓扑来确保每个发送的元组都能得到正确处理。通过跟踪由 Spout 发出的每个元组构成的元组树可以确定元组是否已经完成处理。每个拓扑都有一个“消息延时”参数，如果 Storm 在延时时间内没有检测到元组是否处理完成，就会将该元组标记为处理失败，并会在稍后重新发送该元组。---当任务失败的时候是怎么保证重复执行的,都是放在zookeepr中的这些任务调度数据)

storm woker Executor Task 区别

1个worker进程执行的是1个topology的子集（注：不会出现1个worker为多个topology服务）。

1个worker进程会启动1个或多个executor线程来执行1个topology的component(spout或bolt)。因此，1个运行中的topology就是由集群中多台物理机上的多个worker进程组成的。----如果topology被分到多个work上的话

executor是1个被worker进程启动的单独线程。每个executor只会运行1个topology的1个component(spout或bolt)的task（注：task可以是1个或多个，storm默认是1个component只生成1个task------------------如果设置多个task就会有多个重复的task产生）。

task是最终运行spout或bolt中代码的单元（注：1个task即为spout或bolt的1个实例，executor线程在执行期间会调用该task的nextTuple或execute方法）。topology启动后，1个component(spout或bolt)的task数目是固定不变的，但该component使用的executor线程数可以动态调整（例如：1个executor线程可以执行该component的1个或多个task实例）。这意味着，对于1个component存在这样的条件：#threads<=#tasks（即：线程数小于等于task数目）。默认情况下task的数目等于executor线程数目，即1个executor线程只运行1个task。----这个很重要设置多个就会有多个重复的task

工作原理

Storm集群中有两种节点，一种是控制节点(Nimbus节点)，另一种是工作节点(Supervisor节点)。

所有Topology任务的提交必须在Storm客户端节点上进行(需要配置 storm.yaml文件)，由Nimbus节点分配给其他Supervisor节点进行处理。

Nimbus节点首先将提交的Topology进行分片，分成一个个的Task，并将Task和Supervisor相关的信息提交到 zookeeper集群上

Supervisor会去zookeeper集群上认领自己的Task，通知自己的Worker进程进行Task的处理。

一个Topology运行之后，就会不断的通过Spouts来发送Stream流，通过Bolts来不断的处理接收到的Stream流，Stream流是无界的。 ------------最后一步会不间断的执行，除非手动结束Topology。

Storm不处理计算结果的保存，这是应用代码需要负责的事情，如果数据不大，你可以简单地保存在内存里，也可以每次都更新数据库，也可以采用NoSQL存储。这部分事情完全交给用户。

数据存储之后的展现，也是你需要自己处理的，storm UI 只提供对topology的监控和统计。

Topology

bolt 可以理解为计算程序中的运算或者函数，将一个或者多个数据流作为输入，对数据实施运算后，选择性地输出一个或者多个数据流。 bolt 可以订阅多个由 spout 或者其他bolt 发射的数据流，这样就可以建立复杂的数据流转换网络。而这个网络整体就是拓扑--------------------------即Topology

Steam Group----这个就是负载均衡策略

复杂点的图如下

Shuffle Grouping或None Grouping

随机分组，随机派发stream里面的tuple，下游每个bolt均衡接收到上游的tuple。

Fields Grouping

按字段分组，比如按userid来分组，具有同样userid的tuple会被分到相同的bolt，而不同的userid则被分配到不同的bolots。

All Grouping

广播发送，对于每一个tuple，所有的bolts都会收到。

LocalOrShuffle Grouping

如果下游bolt的某些task与上游spout/bolt的某些task运行在同一个worker进程中，那么上游spout/bolt的这些task所发射的所有tuples均由下游bolt的同进程的tasks来处理；否则，这种分组方式等同于shuffle grouping。

Direct Grouping

直接分组，用这种分组意味着消息的发送者指定优消息接收者的某个task处理这个消息，只有被声明为DirectStream的消息流可以声明这种分组方法。而且这种消息tuple必须使用emitDirect方法来发射。消息处理者可以通过TopologyContext来获取处理它的消息的taskid（OutputCollector.emit方法也会返回taskid）。

集群环境搭建

使用版本2.2.0,在官网上详细说明了需要的配件信息:

搭建Zookeeper

参考:

搭建Python3.x

Python最新的官网下载地址:

#Python3.x 的安装需要依赖这四个组件：gcc， zlib，zlib-devel，openssl-devel；所以需要预先安装[cuiyaonan2000@163.com bin]# yum install gcc -y[cuiyaonan2000@163.com bin]# yum install zlib -y[cuiyaonan2000@163.com bin]# yum install zlib-devel -y[cuiyaonan2000@163.com bin]# yum install openssl-devel -y#下载python3.8.6[cuiyaonan2000@163.com bin]# wget https://www.python.org/ftp/python/3.8.6/Python-3.8.6.tgz#开始编译安装python[cuiyaonan2000@163.com bin]#  cd Python-3.8.6#这个是设置安装的地址[cuiyaonan2000@163.com bin]# ./configure --prefix=/usr/app/python3.8[cuiyaonan2000@163.com bin]#  make && make install#设置环境变量[cuiyaonan2000@163.com bin]# vim  /etc/profile#加入如下的环境变量export PYTHON_HOME=/usr/app/python3.8export  PATH=${PYTHON_HOME}/bin:$PATH#刷新配置文件[cuiyaonan2000@163.com bin]# source /etc/profile#验证是否安装成功[cuiyaonan2000@163.com  Python-3.8.6]# python3Python 3.8.6 (default, Sep 30 2020, 14:33:35) [GCC 4.8.5 20150623 (Red Hat 4.8.5-39)] on linuxType "help", "copyright", "credits" or "license" for more information.>>> eixtTraceback (most recent call last):  File "
   
    ", line 1, in 
    
     NameError: name 'eixt' is not defined>>> exit()

安装了Python3 要检查下yum是否还能使用.可以

vi /usr/bin/yum

测试yum能否正常使用: yum -y install vim

Storm集群环境搭建（1个nimbus+2个supervisor）

首先安装storm

#下载storm[cuiyaonan2000@163.com soft]# wget https://mirrors.bfsu.edu.cn/apache/storm/apache-storm-2.2.0/apache-storm-2.2.0.tar.gz[cuiyaonan2000@163.com soft]# tar -zxvf apache-storm-2.2.0.tar.gz [cuiyaonan2000@163.com soft]# vi /etc/profile#增加如下内容export STORM_HOME=/soft/apache-storm-2.2.0export PATH=$STORM_HOME/bin:$PATH[cuiyaonan2000@163.com soft]# source /etc/profile

修改 /conf/`storm.yaml` 配置文件, 增加如下内容

# Zookeeper集群的主机列表storm.zookeeper.servers:     - "10.1.80.190"     - "10.1.80.191"     - "10.1.80.192"# Nimbus的节点列表 注意这里可以写多个 用于主备用nimbus.seeds: ["10.1.80.190","hadoop002"]# Nimbus和Supervisor需要使用本地磁盘上来存储少量状态（如jar包，配置文件等）storm.local.dir: "/home/storm"# workers进程的端口，每个worker进程会使用一个端口来接收消息#假设只想启动 2 个 worker 的话，此处配置 2 个端口即可。#slot数量一般以每个服务器的CPU线程数来计算。supervisor.slots.ports:     - 6700     - 6701     - 6702     - 6703#修改ui的端口，默认是8080，这个很容易冲突ui.port: 8766#设置work代表的jvm堆空间worker.childopts: "-Xmx2048m"

现在版本交替,配置内容可能不一样,2.2.0的配置项参考:

内容较少----这个版本最好的一点就是Nimbus可以是多个,所以可以进行主备了,当一台挂掉,另一台启动,并且数据不会丢失,因为都存储在zookeeper

启动项目

如果只是nimbus的服务器只需要使用命令nohup storm nimbus &

如果只是supervisor的服务器只需要使用命令nohup storm supervisor &

如果及时nibus有事supervisor 这如上的两个命令都要启动.

Nimbus上运行 ui： nohup storm ui &

Nimbus上运行 logviewer运行： nohup storm logviewer &

# 启动主节点 nimbusnohup storm nimbus &# 启动从节点 supervisor nohup storm supervisor &# 启动UI界面 ui  nohup storm ui &# 启动日志查看服务 logviewer nohup storm logviewer &

启动后可以使用jps来查看程序是否启动成功:

UIServer:即Storm管理界面

Nimbus:即Master

Supervisor:即运算程序