storm 流式计算框架

  • 一:storm 简介
  • 二:storm 的原理与架构
  • 三:storm 的 安装配置
  • 四:storm 的启动脚本

一: storm 的简介:

1.1 storm 是什么:

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

2. 按照storm作者的说法,Storm对于实时计算的意义类似于Hadoop对于批处理的意义。Hadoop提供了map、reduce原语,使我们的批处理程序变得简单和高效。同样,Storm也为实时计算提供了一些简单高效的原语,而且Storm的Trident是基于Storm原语更高级的抽象框架,类似于基于Hadoop的Pig框架,让开发更加便利和高效。本课程会深入、全面的讲解Storm,并穿插企业场景实战讲述Storm的运用。

1.2 实时计算的设计缺点:

数据源务必实时,所以采用Message Queue作为数据源,消息处理Comsumer实时从MQ获取数据进行处理,返回结果到Web或写DB。

这种方式有以下几个缺陷:
1、单机模式,能处理的数据量有限
2、不健壮,服务器挂掉即结束。而Storm集群节点挂掉后,任务会重新分配给其他节点,作业不受影响。
3、失败重试、事务等,你需要在代码上进行控制,过多精力放在业务开发之外。
4、伸缩性差: 当一个消息处理者的消息量达到阀值,你需要对这些数据进行分流, 你需要配置这些新的处理者以让他们处理分流的消息。

1.3 storm 的特点:

1. 适用场景广泛: storm可以实时处理消息和更新DB,对一个数据量进行持续的查询并返回客户端(持续计算),对一个耗资源的查询作实时并行化的处理(分布式方法调用,即DRPC),storm的这些基础API可以满足大量的场景。

2. 可伸缩性高: Storm的可伸缩性可以让storm每秒可以处理的消息量达到很高。扩展一个实时计算任务,你所需要做的就是加机器并且提高这个计算任务的并行度 。Storm使用ZooKeeper来协调集群内的各种配置使得Storm的集群可以很容易的扩展。

3. 保证无数据丢失: 实时系统必须保证所有的数据被成功的处理。 那些会丢失数据的系统的适用场景非常窄, 而storm保证每一条消息都会被处理, 这一点和S4相比有巨大的反差。

4. 异常健壮: storm集群非常容易管理,轮流重启节点不影响应用。

5. 容错性好:在消息处理过程中出现异常, storm会进行重试

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

二:storm 的原理与架构

2.1 Storm集群结构

1. Nimbus 和Supervisors 之间所有的协调工作是通过 一个Zookeeper 集群。

2. Nimbus进程和 Supervisors 进程是无法直接连接和无状态的; 所有的状态维持在Zookeeper中或保存在本地磁盘上。

3. 这意味着你可以 kill -9 Nimbus 或Supervisors 进程,而不需要做备份。
这种设计导致storm集群具有令人难以置信的稳定性,即无耦合。

2.2 storm 的工作原理:

1. Nimbus 负责在集群分发的代码,topo只能在nimbus机器上提交,将任务分配给其他机器,和故障监测。

2. Supervisor,监听分配给它的节点,根据Nimbus 的委派在必要时启动和关闭工作进程。 每个工作进程执行topology 的一个子集。一个运行中的topology 由很多运行在很多机器上的工作进程组成。

3. 在Storm中有对于流stream的抽象,流是一个不间断的无界的连续tuple,注意Storm在建模事件流时,把流中的事件抽象为tuple即元组

4. Storm认为每个stream都有一个源,也就是原始元组的源头,叫做Spout(管口)

5.处理stream内的tuple,抽象为Bolt,bolt可以消费任意数量的输入流,只要将流方向导向该bolt,同时它也可以发送新的流给其他bolt使用,这样一来,只要打开特定的spout再将spout中流出的tuple导向特定的bolt,又bolt对导入的流做处理后再导向其他bolt或者目的地。

可以认为spout就是水龙头,并且每个水龙头里流出的水是不同的,我们想拿到哪种水就拧开哪个水龙头,然后使用管道将水龙头的水导向到一个水处理器(bolt),水处理器处理后再使用管道导向另一个处理器或者存入容器中。

为了增大水处理效率,我们很自然就想到在同个水源处接上多个水龙头并使用多个水处理器,这样就可以提高效率。

这是一张有向无环图,Storm将这个图抽象为Topology(拓扑),Topo就是storm的Job抽象概念,一个拓扑就是一个流转换图

图中每个节点是一个spout或者bolt,每个spout或者bolt发送元组到下一级组件,广播方式。

而Spout到单个Bolt有6种grouping方式

2.3 Topology 作业

Storm将流中元素抽象为tuple,一个tuple就是一个值列表value list,list中的每个value都有一个name,并且该value可以是任意可序列化的类型。拓扑的每个节点都要说明它所发射出的元组的字段的name,其他节点只需要订阅该name就可以接收处理。

2.4 storm 中的角色与概念:

Streams:消息流

消息流是一个没有边界的tuple序列,而这些tuples会被以一种分布式的方式并行创建和处理。 每个tuple可以包含多列,字段类型可以是: integer, long, short, byte, string, double, float, boolean和byte array。 你还可以自定义类型 — 只要你实现对应的序列化器。
Spouts:消息源

Spouts是topology消息生产者。Spout从一个外部源(消息队列)读取数据向topology发出tuple。 消息源Spouts可以是可靠的也可以是不可靠的。一个可靠的消息源可以重新发射一个处理失败的tuple, 一个不可靠的消息源Spouts不会。

Spout类的方法nextTuple不断发射tuple到topology,storm在检测到一个tuple被整个topology成功处理的时候调用ack, 否则调用fail。
storm只对可靠的spout调用ack和fail。

Bolts:消息处理者

消息处理逻辑被封装在bolts里面,Bolts可以做很多事情: 过滤, 聚合, 查询数据库等。
Bolts可以简单的做消息流的传递。复杂的消息流处理往往需要很多步骤, 从而也就需要经过很多Bolts。第一级Bolt的输出可以作为下一级Bolt的输入。而Spout不能有一级。

Bolts的主要方法是execute(死循环)连续处理传入的tuple,成功处理完每一个tuple调用OutputCollector的ack方法,以通知storm这个tuple被处理完成了。当处理失败时,可以调fail方法通知Spout端可以重新发送该tuple。

流程是: Bolts处理一个输入tuple, 然后调用ack通知storm自己已经处理过这个tuple了。storm提供了一个IBasicBolt会自动调用ack。
Bolts使用OutputCollector来发射tuple到下一级Blot。

三:storm 的 安装配置

3.1 安装storm环境准备

3.1.1 安装zookeeper服务:

tar -zxvf zookeeper-3.4.5.tar.gz
mv zookeeper-3.4.5 /usr/local/zookeeper

---
cd /usr/local/zookeeper
mkdir data
cd data
echo "1" > myid
--
cd /usr/local/zookeeper/conf

cp -p zoo_sample.cfg zoo.cfg
vim zoo.cfg

dataDir=/usr/local/zookeeper/data
server.1=master:2888:3888
server.2=slave1:2888:3888
server.3=slave2:2888:3888
---

cd /usr/local/
tar -zcvf zookeeper.tar.gz zookeeper

--- 同步到slave1 与slave2 节点----

scp zookeeper.tar.gz [email protected]:/usr/local/
scp zookeeper.tar.gz [email protected]:/usr/local/

------------------slave1 节点---------------------
cd /usr/local/
tar -zxvf zookeeepr.tar.gz 

cd zookeeper/data
echo ‘2‘ > myid

------------------slave2 节点--------------------
cd /usr/local/
tar -zxvf zookeeepr.tar.gz 

cd zookeeper/data
echo ‘3‘ > myid

3.1.2 zookeeper 的启动脚本范例:

#!/bin/bash

if [ $# -ne 1 ]; then
        echo "Usage: sh start_zookeeper.sh [start|status|stop]"
        exit 2
fi

for node in master slave1 slave2     # ---这个地方有多少个主机就加多少
do
        echo "$1 in $node"
        ssh $node "source /etc/profile && /opt/modules/zookeeper-3.4.5/bin/zkServer.sh $1"
done

3.1.3 安装依赖包:

1. CentOS6.4 安装相关编译工具包

yum install -y gcc gcc++* gcc-c++ uuid-devel libuuid-devel libtool git
2. 安装 ZeroMQ
wget http://download.zeromq.org/zeromq-2.1.7.tar.gz
tar -xzvf zeromq-2.1.7.tar.gz
cd zeromq-2.1.7
./configure
make
make install

3. JZMQ安装

git clone https://github.com/nathanmarz/jzmq.git
cd jzmq
./autogen.sh
./configure
make
make install

3.1.4 storm 安装:

上传文件apache-storm-0.9.0.6.tar.gz 到/home/hadoop下面

cd /usr/local/storm
mkdir data
cd conf
---
vim storm.yaml

########### These MUST be filled in for a storm configuration
 storm.zookeeper.servers:
     - "master"
     - "slave1"
     - "slave2"
#
 nimbus.host: "master"
# 

---

## Locations of the drpc servers
 drpc.servers:
     - "master"
     - "slave1"
     - "slave2"
---
  增加storm 任务的目录与端口:
---
 storm.local.dir: "/usr/local/storm/data"
 supervisor.slots.ports:
 - 6701
 - 6702
 - 6703
 - 6704
ui.port: 8081

------同步所有节点----------
cd /usr/local/
tar -zcvf storm.tar.gz storm

scp storm.tar.gz [email protected]:/usr/local/
scp storm.tar.gz [email protected]:/usr/local/
---------------slave1节点----------

tar -zxvf storm.tar.gz 

---------------slave2节点----------

tar -zxvf storm.tar.gz


3.1.4 启动服务与浏览器访问

1. 启动zookeeper 服务

master:
 cd /usr/local/zookeeper/
 bin/zkServer.sh start
 --------------------------------------
 slave1:
 cd /usr/local/zookeeper/
 bin/zkServer.sh start
 --------------------------------------
 slave2:
 cd /usr/local/zookeeper/
 bin/zkServer.sh start 

 --------------------------------------
 2. 启动storm的相关服务

 master:

 cd /usr/local/storm/
 bin/storm nimbus &
 bin/storm ui &
 ------------------------------
 slave1

 cd /usr/local/storm/
 bin/storm supervisor & 

 ------------------------------
 slave2
 cd /usr/local/storm/
 bin/storm supervisor &  

 -------------------------------

 打开浏览器访问:

 http://192.168.3.1:8081

运行wordcount 实例:

bin/storm jar examples/storm-starter/storm-starter-topologies-0.9.6.jar storm.starter.WordCountTopology wordcount

3.1.5 storm 的参数解析

1. storm.zookeeper.servers:这是一个为Storm集群配置的Zookeeper集群的主机列表

2. storm.local.dir:Nimbus和Supervisor守护程序需要一个本地磁盘目录存储小量状态(像jars,confs,其它),每台机器都创建这些目录,赋可写权限

3. java.library.path:这是Storm使用的本地库(ZeroMQ和JZMQ)载入路径。大多数安装,默认路径"/usr/local/lib:/opt/local/lib:/usr/lib"就行,所以你可能不需要配置它。

4. supervisor.slots.ports:?每一台worker机器,你用这个配置来指定多少workers运行在那台机。每个worker使用单一端口接收消息,并且这个设置定义哪个端口是打开的且可以使用。如果你定义5个端口,那么Storm将在这台机分配5个worker运行。

------------------------------------

Storm是一个快速失败(fail-fast)的系统,这意味着这些进程随时都可能因发生错误而停止。由于Storm的设计,所以它随时停止都是安全的,当进程重新启动时正确的恢复。这是为什么Storm保持进程无状态的原因--?如果Nimbus或supervisors重新启动,正在运行的topologies是不受影响的。

Nohup挂到后台执行
1)Nimbus
在master机器的supervision下运行命令”bin/storm nimbus”
2)Supervisor
在每个worker机器的supervision下运行命令”bin/storm supervisor”。Supervisor守护程序负责starting 和 stopping 那台机上的worker进程
3)UI
运行supervision下的命令”bin/storm ui”来运行Storm UI(你能从浏览器访问一个站点,它提供集群和topologies的诊断信息)。在你的浏览器中输入”?http://{nimbus host}:8081”访问UI。

3.1.6 strom 的nimbus的主节点特点:

nimbus topology任务提交后,程序是运行在supervisor节点上
        Nimbus不参与程序的运行

        Nimbus出现故障,不能提交Topology,已经提交了的Topology还是
        正常运行在集群上

        已经运行在集群上Topology,如果这时候某些task出现异常
        则无法重现分配节点

-----------------------------------------------------------------
        查看Topology运行日志:
            需要启动一个进程 logviewer
            需要在每个supervisor节点上启动,不用在nimbus节点上启动
            bin/storm logviewer > ./logs/logviewer.out 2>&1 &

        nimbus supervisor ui logviewer

3.1.7 停掉storm 的 worldcount 程序

  activate 激活
  deactivate 暂停
  Repalance 从新分配
  kill 杀掉这个 toplogy

直接通过命令行执行:# bin/storm kill wordcount(提交的时候
    指定的Topology名称)


3.1.8 先看下Zookeeper怎么存储相关状态信息的

登录到zookeeper 的里面去

cd /usr/local/zookeeper/bin

./zkCli.sh

znode:
    /storm
        /workerbeats  worker心跳信息
        /errors       topology运行过程中Task运行异常信息
        /supervisors  记录supervisor状态心跳信息
        /storms       记录的是topology任务信息
        /assignments  记录的是Topology任务的分配信息

3.1.9 nimbus supervisor ui logviewer进程的关闭

     kill -9 `ps -ef | grep daemon.nimbus | awk ‘{print $2}‘ | head -n 1`
     kill -9 `ps -ef | grep ui.core | awk ‘{print $2}‘ | head -n 1`
     kill -9 `ps -ef | grep daemon.supervisor | awk ‘{print $2}‘ | head -n 1`
     kill -9 `ps -ef | grep daemon.logviewer | awk ‘{print $2}‘ | head -n 1`

3.2.0 storm 的启动脚本

#!/bin/bash

source /etc/profile

STORM_HOME=/opt/modules/apache-storm-0.9.6
## 主节点  nimbus ui
${STORM_HOME}/bin/storm nimbus > /dev/null 2>&1 &
${STORM_HOME}/bin/storm ui > /dev/null 2>&1 &

## 从节点  supervisor  logviewer
for supervisor in `cat ${STORM_HOME}/bin/stormSupervisorHosts`
do
        echo "start supervisor and logviewer in $supervisor"
        ssh $supervisor "source /etc/profile && ${STORM_HOME}/bin/storm supervisor > /dev/null 2>&1 &" &
        ssh $supervisor "source /etc/profile && ${STORM_HOME}/bin/storm logviewer > /dev/null 2>&1 &" &
done

3.2.1 storm 的停止脚本

#!/bin/bash

source /etc/profile
STORM_HOME=/opt/modules/apache-storm-0.9.6
### 主节点 nimbus ui
kill -9 `ps -ef | grep daemon.nimbus | awk ‘{print $2}‘ | head -n 1`
kill -9 `ps -ef | grep ui.core | awk ‘{print $2}‘ | head -n 1`

### 从节点 supervisor logviewer
for supervisor in `cat ${STORM_HOME}/bin/stormSupervisorHosts`
do
        echo "stop supervisor and logviewer in $supervisor"
        ssh $supervisor kill -9 `ssh $supervisor "ps -ef| grep daemon.supervisor| awk ‘{print $2}‘ | head -n 1" ` > /dev/null 2>&1 &
        ssh $supervisor kill -9 `ssh $supervisor "ps -ef| grep daemon.logviewer| awk ‘{print $2}‘ | head -n 1" ` >/dev/null 2>&1 &

done

原文地址:http://blog.51cto.com/flyfish225/2097505

时间: 2024-08-03 17:32:07

storm 流式计算框架的相关文章

流式计算框架-STORM简介

在当前的数据分析领域,对实时数据的计算需求越来越强烈,在此领域,出现了各类计算框架,如:Storm.S4等.目前本土公司对这些流式计算框架的应用也比较广泛,但苦于相关文档英文居多,缺少成系列且与官方相对应的中文手册.本系列试图从官方文档翻译入手,给大家呈现较为完备的中文资料,同时也是对自身知识的总结沉淀. 在这个系列博客中,我们选择了twitter的Storm框架,原因很简单,因为本人长期使用的就是该框架,咱们先从简介开始. Apache Storm是一个免费.开源.分布式的实时计算系统.相对于

大数据技术(1)流式计算与Storm

2011年在海量数据处理领域,Hadoop是人们津津乐道的技术,Hadoop不仅可以用来存储海量数据,还以用来计算海量数据.因为其高吞吐.高可靠等特点,很多互联网公司都已经使用Hadoop来构建数据仓库,高频使用并促进了Hadoop生态圈的各项技术的发展.一般来讲,根据业务需求,数据的处理可以分为离线处理和实时处理,在离线处理方面Hadoop提供了很好的解决方案,但是针对海量数据的实时处理却一直没有比较好的解决方案. 就在人们翘首以待的时间节点,storm横空出世,与生俱来的分布式.高可靠.高吞

什么是流式计算?

一.流式计算的背景 在日常生活中,我们通常会先把数据存储在一张表中,然后再进行加工.分析,这里就涉及到一个时效性的问题.如果我们处理以年.月为单位的级别的数据,那么多数据的实时性要求并不高:但如果我们处理的是以天.小时,甚至分钟为单位的数据,那么对数据的时效性要求就比较高.在第二种场景下,如果我们仍旧采用传统的数据处理方式,统一收集数据,存储到数据库中,之后在进行分析,就可能无法满足时效性的要求. 二.流式计算与批量计算 大数据的计算模式主要分为批量计算(batch computing).流式计

流式计算(一)-Java8Stream

大约各位看官君多少也听说了Storm/Spark/Flink,这些都是大数据流式处理框架.如果一条手机组装流水线上不同的人做不同的事,有的装电池,有的装屏幕,直到最后完成,这就是典型的流式处理.如果手机组装是先全部装完电池,再交给装屏幕的组,直到完成,这就是旧式的集合式处理.今天,就来先说说JDK8中的流,虽然不是很个特新鲜的话题,但是一个很好的开始,因为——思想往往比细节重要! 准备: Idea2019.03/Gradle5.6.2/JDK11.0.4/Lambda 难度:新手--战士--老兵

Strom流式计算

序言 主要学习方向 Kafka 分布式消息系统 Redis 缓存数据库 Storm 流式计算 1.Storm 的基本概念 2.Storm 的应用场景 3.Storm 和Hadoop的对比 4.Storm 集群的安装的linux环境准备 5.zookeeper集群搭建 6.Storm 集群搭建 7.Storm 配置文件配置项讲解 8.集群搭建常见问题解决 9.Storm 常用组件和编程 API:Topology. Spout.Bolt 10.Storm分组策略(stream groupings)

storm:最火的流式处理框架

本文出处:www.cnblogs.com/langtianya/p/5199529.html 伴随着信息科技日新月异的发展,信息呈现出爆发式的膨胀,人们获取信息的途径也更加多样.更加便捷,同时对于信息的时效性要求也越来越高.举个搜索场景中的例子,当一个卖家发布了一条宝贝信息时,他希望的当然是这个宝贝马上就可以被卖家搜索出来.点击.购买啦,相反,如果这个宝贝要等到第二天或者更久才可以被搜出来,估计这个大哥就要骂娘了.再举一个推荐的例子,如果用户昨天在淘宝上买了一双袜子,今天想买一副泳镜去游泳,但是

流式处理框架storm浅析

前言前一段时间参与哨兵流式监控功能设计,调研了两个可以做流式计算的框架:storm和spark streaming,我负责storm的调研工作.断断续续花了一周的时间看了官网上的doc和网络上的一些资料.我把所学到的总结成一个文档,发出来给对storm感兴趣的同事做入门引导. storm背景随着互联网的更进一步发展,从Portal信息浏览型到Search信息搜索型到SNS关系交互传递型,以及电子商务.互联网旅游生活产品等将生活中的流通环节在线化.对效率的要求让大家对于实时性的要求进一步提升,而信

流式处理框架对比

分布式流处理是对无边界数据集进行连续不断的处理.聚合和分析的过程,与MapReduce一样是一种通用计算框架,期望延迟在毫秒或者秒级别.这类系统一般采用有向无环图(DAG).DAG是任务链的图形化表示,用它来描述流处理作业的拓扑.在选择不同的流处理系统时,通常会关注以下几点: 运行时和编程模型:平台框架提供的编程模型决定了许多特色功能,编程模型要足够处理各种应用场景. 函数式原语:流处理平台应该能提供丰富的功能函数,比如,map或者filter这类易扩展.处理单条信息的函数;处理多条信息的函数a

大数据读书笔记(2)-流式计算

早期和当前的"流式计算"系统分别称为"连续查询处理类"和"可扩展数据流平台类"计算系统. 流式计算系统的特点: 1)低延迟 2)极佳的系统容错性 3)极强的系统扩展能力 4)灵活强大的应用逻辑表达能力 目前典型的流式计算系统: S4,storm,millwheel,samza,d-stream,hadoop online,mupd8等. 其中storm和millwheel是各方面比较突出的. 流式计算系统架构: 常见的流式计算系统架构分为两种:主