首页 > 学院 > 开发设计 > 正文

Storm入门学习随记

2019-11-14 09:58:25
字体:
来源:转载
供稿:网友

推荐慕课网视频:http://www.imooc.com/video/10055

 

====Storm的起源。

Storm是开源的、分布式、流式计算系统

 

什么是分布式呢?就是将一个任务拆解给多个计算机去执行,让许多机器共通完成同一个任务,

把这个多机的细节给屏蔽,对外提供同一个接口、同一个服务,这样的系统就是分布式系统。

 

在多年以前并没有非常范用的分布式系统,即使存在,也都是限定在指定的领域,

当然,也有人尝试从中提取出共通的部分,发明一个通用的分布式系统,但是都没有很好的结果。

后来,Google发表了3篇论文,提出了分布式计算的模型,在分布式系统上有了一个质的突破。

 

有一位大牛看了这3篇论文之后,深受启发,然后就发明了Hadoop系统。

 

然后,基于Hadoop的改造系统就如雨后春笋一般,接二连三的出现了。

以至于,Hadoop已经不是一套软件,而是一整套生态系统了。

于是,人们谈到分布式,就必谈Hadoop了。

 

但是,Hadoop并不是万能的,它只能处理适合进行批量计算的需求。对于,非批量的计算就不能够满足要求了。

很多时候,我们只能先收集一段时间数据,等数据收集到一定规模之后,我们才开始MaPReduce处理。

 

有这么一个故事:

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

路人甲是在一家媒体公司A工作,他的主要工作内容很简单,就是在一些搜索引擎上做广告,

众所周知,搜索引擎上的广告是竞价排名的,谁土豪谁就排前面,出钱少的就只能排在后面。

公司A的竞争对手都比较土豪,所以呢,公司A的广告就一直排在后面,也没什么好的办法。

后来,路人甲想出了一个馊主意,就是用程序不断的去点击竞争对手的广告,让对手的广告费

很快的花费调,这样公司A就可以廉价的将广告排在前面了。

搜索引擎公司试图识别出这些恶意点击屏来保护商家,将这些恶意点击扣除的费用返还给商家。

一般来说呢,如果利用MapReduce,一般情况下,都需要收集一段时间数据,然后根据这些

数据来算出哪些点击是恶意的,本身收集数据就已经很耗费时间了,再等计算完毕之后,

土豪商家的广告费也基本上不剩什么了。

所以呢,我们希望在点击发生的时候就算出来该点击是否是作弊行为,及时不能马上判断出,

也应该尽早的计算出来。

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

 为了解决上面这个故事的需求,分布式流式计算系统就产生了,比较知名的有:

•【Yahoo】S4

•【IBM】StreamBase

•【Amazon】Kinesis

•【Spark】Streaming

•【Google】Millwheel

•【Apache】Storm(目前业界中最知名、流程)

 

批量计算(以Hadoop为代表)与流式计算的区别有哪些呢?

 

###################

目前已经有人在做一些前瞻性的项目,这些人试图将批量计算和流式计算进行整合

试图使用同一套API,即搞定流式计算,又搞定批量计算。

使一段代码不要任何改动,就可以同时执行在批量计算和流式计算两种系统之上。

这种系统目前比较有名的有:

【Twitter】Summing Bird

【Google】CloudDataflow

两个接口都已经开源了。等以后有机会一定要提前接触一下。

###################

 

====Storm组件

Storm采用的是主从结构,就是使用一台主节点来管理整个集群的运行状态。

这个主节点被称为:Nimbus,从节点用来维护每台机器的状态,被称为:Supervisor

 

为什么采取主从结构呢?主从结构比较简单,不需要进行主节点仲裁等工作。

 

从前面的结构图中我们还可以看出,采取主从结构之后,Nimbus是一个单点,

但是,我们知道分布式领域里,大家都比较讨厌自己的系统设计中存在单点,

因为单点如果发生故障,很有可能影响到整个集群的可用性。

 

所以,如果一个系统设计中如果存在单点,一般情况下这个单点的作业必然比较轻,

挂了之后,短时间之内也不影响真个系统的运行,并且一般情况下都是没有状态的,

宕机之后至需要重启就能够恢复并正确处理。

 

Nimbus的角色是只负责一些管理性的工作,它并不关心Worker之间的数据是如何传输的,

它的一些主要状态都存在分布式协调服务(Zookeeper)中,内存里面的东西都是可以丢失的,

如果它挂掉,只要没有运算节点发生故障,那么整个作业还是能够正常的进行数据处理的。

Nimbus重启之后,就可以正确处理真个系统的事务了。

 

Supervisor的角色是听Nimbus的话,来启动并监控真正进行计算的Worker的进程,

如果Worker有异常,那么久帮助Worker重启一下,它也不负责数据计算和数据传输,

 

真正的数据计算和输出,都是由Worker来进行。

Worker是运行在工作节点上面,被Supervisor守护进程创建的用来干活的JVM进程。

每个Worker对应于一个给定topology的全部执行任务的一个子集。

反过来说,一个Worker里面不会运行属于不同的topology的执行任务。

 

====Storm UI

为了方便用户管理集群,查看集群运行状态,提供了一个基于Web的UI来监控整个Storm集群

它本身不是集群运行的必须部分,它的启动停止都不影响Storm的正常运行。

 

====Storm作业提交运行流程

(1)用户使用Storm的API来编写Storm Topology。

(2)使用Storm的Client将Topology提交给Nimbus。

Nimbus收到之后,会将把这些Topology分配给足够的Supervisor。

(3)Supervisor收到这些Topoligy之后,Nimbus会指派一些Task给这些Supervisor。

(4)Nimvus会指示Supervisor为这些Task生成一些Worker。

(5)Worker来执行这些Task来完成计算任务。

  

====StormAPI基础概念

Storm称用户的一个作业为Topology(拓扑)。

 

为什么叫拓扑呢?是因为Storm的一个拓扑主要包含了许多的数据节点,还有一些计算节点,

以及这些节点之间的边,也就是说Storm的拓扑是由这些点和边组成的一个有向无环图。

这些点有两种:数据源节点(Spout)、普通的计算节点(Bolt),

点之间的边称为数据流(Stream),数据流中的每一条记录称为Tuple。

 

如下图中,每一个“水龙头”表示一个Spout,它会发送一些Tuple给下游的Bolt,

这些Bolt经过处理周,再发送一个Tuple给下一个Bolt,

最后,在这些Bolt里面是可以执行一些写数据到外部存储(如数据库)等操作的。

在图中这个Topology里面我们看到了两个Spout和5个Bolt,

在实际运行的时候,每个Spout节点都可能有很多个实例,每个Bolt也有可能有很多个实例。

就像MapReduce一样,一个Map节点并不代表只有一个并发,而有可能很多个Map实例在跑。

 

这些Spout和Bolt的这些边里面,用户可以设置多种的Grouping的方式。

有些类似SQL中的Group By。用来制定这些计算是怎么分组的。

 

*Fields Grouping:保证同样的字段移动落到同一个Bolt里。

 

--以WordCount为例,MapReduce和Storm的工作流程对比:

(1)MapReduce

(2)Storm

 

====各个组件的一些说明

--Topologies

为了在storm上面做实时计算, 你要去建立一些topologies。一个topology就是一个计算节点所组成的图。Topology里面的每个处理节点都包含处理逻辑, 而节点之间的连接则表示数据流动的方向。运行一个Topology是很简单的。首先,把你所有的代码以及所依赖的jar打进一个jar包。然后运行类似下面的这个命令。

strom jar all-your-code.jar backtype.storm.MyTopology arg1 arg2

这个命令会运行主类: backtype.strom.MyTopology,参数是arg1, arg2。这个类的main函数定义这个topology并且把它提交给Nimbus。storm jar负责连接到nimbus并且上传jar文件。

 

--Stream

Stream是storm里面的关键抽象。一个stream是一个没有边界的tuple序列。storm提供一些原语来分布式地、可靠地把一个stream传输进一个新的stream。比如: 你可以把一个tweets流传输到热门话题的流。storm提供的最基本的处理stream的原语是spout和bolt。你可以实现Spout和Bolt对应的接口以处理你的应用的逻辑。spout是流的源头。比如一个spout可能从Kestrel队列里面读取消息并且把这些消息发射成一个流。又比如一个spout可以调用twitter的一个api并且把返回的tweets发射成一个流。通常Spout会从外部数据源(队列、数据库等)读取数据,然后封装成Tuple形式,之后发送到Stream中。Spout是一个主动的角色,在接口内部有个nextTuple函数,Storm框架会不停的调用该函数。

 

bolt可以接收任意多个输入stream, 作一些处理, 有些bolt可能还会发射一些新的stream。一些复杂的流转换, 比如从一些tweet里面计算出热门话题, 需要多个步骤, 从而也就需要多个bolt。Bolt可以做任何事情: 运行函数,过滤tuple,做一些聚合,做一些合并以及访问数据库等等。Bolt处理输入的Stream,并产生新的输出Stream。Bolt可以执行过滤、函数操作、Join、操作数据库等任何操作。Bolt是一个被动的角色,其接口中有一个execute(Tuple input)方法,在接收到消息之后会调用此函数,用户可以在此方法中执行自己的处理逻辑。

 

spout和bolt所组成一个网络会被打包成topology, topology是storm里面最高一级的抽象(类似 Job), 你可以把topology提交给storm的集群来运行。topology的结构在Topology那一段已经说过了,这里就不再赘述了。

 

topology里面的每一个节点都是并行运行的。 在你的topology里面, 你可以指定每个节点的并行度, storm则会在集群里面分配那么多线程来同时计算。一个topology会一直运行直到你显式停止它。storm自动重新分配一些运行失败的任务, 并且storm保证你不会有数据丢失, 即使在一些机器意外停机并且消息被丢掉的情况下。

 

--数据模型(Data Model)

storm使用tuple来作为它的数据模型。每个tuple是一堆值,每个值有一个名字,并且每个值可以是任何类型,在我的理解里面一个tuple可以看作一个没有方法的java对象(或者是一个表的字段)。总体来看,storm支持所有的基本类型、字符串以及字节数组作为tuple的值类型。你也可以使用你自己定义的类型来作为值类型, 只要你实现对应的序列化器(serializer)。一个Tuple代表数据流中的一个基本的处理单元,例如一条cookie日志,它可以包含多个Field,每个Field表示一个属性。

 

Tuple本来应该是一个Key-Value的Map,由于各个组件间传递的tuple的字段名称已经事先定义好了,所以Tuple只需要按序填入各个Value,所以就是一个Value List。一个没有边界的、源源不断的、连续的Tuple序列就组成了Stream。

 

topology里面的每个节点必须定义它要发射的tuple的每个字段。比如下面这个bolt定义它所发射的tuple包含两个字段,类型分别是: double和triple。public class DoubleAndTripleBoltimplementsIRichBolt {    private OutputCollectorBase _collector;    @Override    public void prepare(Map conf, TopologyContext context, OutputCollectorBase collector) {        _collector = collector;    }    @Override    public void execute(Tuple input) {        intval = input.getInteger(0);        _collector.emit(input,newValues(val*2, val*3));        _collector.ack(input);    }    @Override    public void cleanup() {    }    @Override    public void declareOutputFields(OutputFieldsDeclarer declarer) {        declarer.declare(newFields("double","triple"));    }}

 

参考博客:http://blog.itpub.net/29754888/viewspace-1260026/

 

====StormAPI使用

我们来看看WorldCount的example代码。

 

====Storm的并发机制

Task数量:表示每个Spout或Bolt逻辑上有多少个并发。它影响输出结果。

Worker数量:代表总共有几个JVM进程去执行我们的作业。

Executor数量:表示每个Spout或Bolt启动几个线程来运行。

 

下面代码中的数字表示Executor数量,它不影响结果,影响性能。

 

Worker的数量在Config中设置,下图代码中的部分表示Worker数量。

*本地模式中,Worker数不生效,只会启动一个JVM进行来执行作业。

*只有在集群模式设置Worker才有效。而且集群模式的时候一定要设置才能体现集群的价值。

 

====Storm数据可靠性

分布式系统都管理很多台机器,需要保证任意的Worker挂掉之后,我们的系统仍然能正确的处理,那么

Storm如何保证这些数据正确的恢复?

Storm如何保证这些数据不被重复计算?

(1)Spout容错API:NextTuple中,emit时,指定MsgID。

(2)Bolt容错API:①emit时,锚定输入Tuple。②Act输入Tuple。

 

====Storm集群搭建

(1)安装zookeeper集群

配置方法省略。

 

(2)下载安装Storm

官网上下载Storm:http://storm.apache.org

上传至linux并解压缩。这里将Storm解压缩到/opt/apache-storm-0.10.0路径下了。

 

(3)修改Storm配置文件

配置文件路径:/opt/apache-storm-0.9.5/conf/storm.yaml

配置内容如下:

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

storm.zookeeper.servers:- "192.168.93.128"- "192.168.93.129"- "192.169.93.130"nimbus.host: "192.168.93.128"storm.local.dir: "/opt/apache-storm-0.9.5/status"supervisor.slots.ports:- 6700- 6701- 6702- 6703

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

 

置之后的文件如下如所示:

 

--Storm配置项详细介绍

•storm.zookeeper.servers:

ZooKeeper服务器列表

•storm.zookeeper.port:

ZooKeeper连接端口

•storm.local.dir:

storm使用的本地文件系统目录(必须存在并且storm进程可读写)

•storm.cluster.mode:

Storm集群运行模式([distributed|local])

•storm.local.mode.zmq:

Local模式下是否使用ZeroMQ作消息系统,如果设置为false则使用java消息系统。默认为false

•storm.zookeeper.root:

ZooKeeper中Storm的根目录位置

•storm.zookeeper.session.timeout:

客户端连接ZooKeeper超时时间

•storm.id:

运行中拓扑的id,由storm name和一个唯一随机数组成。

•nimbus.host:

nimbus服务器地址

•nimbus.thrift.port:nimbus的thrift监听端口

•nimbus.childopts:

通过storm-deploy项目部署时指定给nimbus进程的jvm选项

•nimbus.task.timeout.secs:

心跳超时时间,超时后nimbus会认为task死掉并重分配给另一个地址

•nimbus.monitor.freq.secs:

nimbus检查心跳和重分配任务的时间间隔。注意如果是机器宕掉nimbus会立即接管并处理

•nimbus.supervisor.timeout.secs:

supervisor的心跳超时时间,一旦超过nimbus会认为该supervisor已死并停止为它分发新任务

•nimbus.task.launch.secs:

task启动时的一个特殊超时设置。在启动后第一次心跳前会使用该值来临时替代nimbus.task.timeout.secs

•nimbus.reassign:

当发现task失败时nimbus是否重新分配执行。默认为真,不建议修改

•nimbus.file.copy.expiration.secs:

nimbus判断上传/下载链接的超时时间,当空闲时间超过该设定时nimbus会认为链接死掉并主动断开

•ui.port:

Storm UI的服务端口

•drpc.servers:

DRPC服务器列表,以便DRPCSpout知道和谁通讯

•drpc.port:

Storm DRPC的服务端口

•supervisor.slots.ports:

supervisor上能够运行workers的端口列表。每个worker占用一个端口,且每个端口只运行一个worker。

通过这项配置可以调整每台机器上运行的worker数。(调整slot数/每机)

•supervisor.childopts:

在storm-deploy项目中使用,用来配置supervisor守护进程的jvm选项

•supervisor.worker.timeout.secs:

supervisor中的worker心跳超时时间,一旦超时supervisor会尝试重启worker进程.

•supervisor.worker.start.timeout.secs:

supervisor初始启动时,worker的心跳超时时间,当超过该时间supervisor会尝试重启worker。

因为JVM初始启动和配置会带来的额外消耗,从而使得第一次心跳会超过supervisor.worker.timeout.secs的设定

•supervisor.enable:

supervisor是否应当运行分配给他的workers。默认为true,该选项用来进行Storm的单元测试,一般不应修改.

•supervisor.heartbeat.frequency.secs:

supervisor心跳发送频率(多久发送一次)

•supervisor.monitor.frequency.secs:

supervisor检查worker心跳的频率

•worker.childopts:

supervisor启动worker时使用的jvm选项。所有的”%ID%”字串会被替换为对应worker的标识符

•worker.heartbeat.frequency.secs:

worker的心跳发送时间间隔

•task.heartbeat.frequency.secs:

task汇报状态心跳时间间隔

•task.refresh.poll.secs:

task与其他tasks之间链接同步的频率。(如果task被重分配,其他tasks向它发送消息需要刷新连接)

。一般来讲,重分配发生时其他tasks会理解得到通知。该配置仅仅为了防止未通知的情况。

•topology.debug:

如果设置成true,Storm将记录发射的每条信息。

•topology.optimize:

master是否在合适时机通过在单个线程内运行多个task以达到优化topologies的目的

•topology.workers:

执行该topology集群中应当启动的进程数量。

每个进程内部将以线程方式执行一定数目的tasks。topology的组件结合该参数和并行度提示来优化性能

•topology.ackers:

topology中启动的acker任务数。

Acker保存由spout发送的tuples的记录,并探测tuple何时被完全处理。

当Acker探测到tuple被处理完毕时会向spout发送确认信息。通常应当根据topology的吞吐量来确定acker的数目,但一般不需要太多。

当设置为0时,相当于禁用了消息可靠性。storm会在spout发送tuples后立即进行确认

•topology.message.timeout.secs:

topology中spout发送消息的最大处理超时时间。

如果一条消息在该时间窗口内未被成功ack,Storm会告知spout这条消息失败。而部分spout实现了失败消息重播功能。

•topology.kryo.register:

注册到Kryo(Storm底层的序列化框架)的序列化方案列表。序列化方案可以是一个类名,或者是com.esotericsoftware.kryo.Serializer的实现

•topology.skip.missing.kryo.registrations:

Storm是否应该跳过它不能识别的kryo序列化方案。如果设置为否task可能会装载失败或者在运行时抛出错误

•topology.max.task.parallelism:

在一个topology中能够允许的最大组件并行度。该项配置主要用在本地模式中测试线程数限制.

•topology.max.spout.pending:

一个spout task中处于pending状态的最大的tuples数量。该配置应用于单个task,而不是整个spouts或topology

•topology.state.synchronization.timeout.secs:

组件同步状态源的最大超时时间(保留选项,暂未使用)

•topology.stats.sample.rate:

用来产生task统计信息的tuples抽样百分比

•topology.fall.back.on.java.serialization:

topology中是否使用java的序列化方案

•zmq.threads:

每个worker进程内zeromq通讯用到的线程数

•zmq.linger.millis:

当连接关闭时,链接尝试重新发送消息到目标主机的持续时长。这是一个不常用的高级选项,基本上可以忽略.

•java.library.path:

JVM启动(如Nimbus,Supervisor和workers)时的java.library.path设置。该选项告诉JVM在哪些路径下定位本地库

 

(4)配置Storm环境变量

环境变量位置:/etc/profile

配置内容之后如下图所示:

注意:环境变量修改只有,一定要使用Source命令来使之生效。

 

(5)启动Storm

--启动Storm UI

命令:storm ui >/dev/null 2>&1 &

我们可以它启动的时候相关的输出指向到/def/null,并且把错误也重新定向到正常输出。

 

--启动主节点(Nimbus节点)

命令:storm nimbus >/dev/null 2>&1 &

在第1台Linux虚拟机上执行。正常启动时的jps结果如下图所示:

 

--启动工作节点(Supervisor节点)

命令:storm supervisor >/dev/null 2>&1 &

在第2、3台Linux虚拟机上执行。正常启动时的jps结果如下图所示:

 

(6)启动StormUI监控页面:

Storm正常启动之后,应该可以打开StormUI画面。在浏览器中输入地址和端口即可

正确启动时应该如下图所示:

--Mainpage:

main页面主要包括3个部分

 

 【Cluster Summary】

•Nimbus uptime: nimbus的启动时间

•Supervisors: storm集群中supervisor的数目

•used slots: 使用了的slots数

•free slots: 剩余的slots数

•total slots: 总的slots数

•Running tasks: 运行的任务数

 

【topology summary】

•Name: topology name

•id: topology id (由storm生成)

•status: topology的状态,包括(ACTIVE, INACTIVE, KILLED, REBALANCING)

•uptime: topology运行的时间

•num workers: 运行的workers数

•num tasks: 运行的task数

 

【supervisor summary】

•host: supervisor(主机)的主机名

•uptime: supervisor启动的时间

•slots: supervisor的端口数

•used slots: 使用的端口数

 

--Topology page

topology页面主要包括4个部分

【topology summary】

(同主页)

 

【topology stats】

•window: 时间窗口,显示10m、3h、1d和all time的运行状况

•emitted: emitted tuple数

•transferred: transferred tuple数, 说下与emitted的区别:如果一个task,emitted一个tuple到2个task中,则transferred tuple数是emitted tuple数的两倍

•complete latency: spout emitting 一个tuple到spout ack这个tuple的平均时间

•acked: ack tuple数

•failed: 失败的tuple数

 

【spouts】

•id: spout id

•parallelism: 任务数

•last error: 最近的错误数,只显示最近的前200个错误

•emitted、transferred、complete latency、acked和failed上面已解释

 

【bolts】

•process latency: bolt收到一个tuple到bolt ack这个tuple的平均时间

其他参数都解释过了

还有componentpage和taskpage,参数的解释同上。

taskpage中的Component指的是spoutid或者boltid,time指的是错误发生的时间,error是指错误的具体内容。

 

====Storm常用命令

【提交Topologies】

命令格式:storm jar 【jar路径】 【拓扑包名.拓扑类名】 【拓扑名称】

样例:storm jar /storm-starter.jar storm.starter.WordCountTopology wordcountTop

#提交storm-starter.jar到远程集群,并启动wordcountTop拓扑。

 

【停止Topologies】

命令格式:storm kill 【拓扑名称】

样例:storm kill wordcountTop

#杀掉wordcountTop拓扑。

 

【启动nimbus后台程序】

命令格式:storm nimbus

 

【启动supervisor后台程序】

命令格式:storm supervisor

 

【启动drpc服务】

命令格式:storm drpc

 

【启动ui服务】

命令格式:storm ui

 

【启动REPL】

REPL — read-evaluate-print-loop。

虽然clojure可以作为一种脚本语言内嵌在java里面,但是它的首选编程方式是使用REPL,这是一个简单的命令行接口,

使用它你可以输入你的命令,执行,然后查看结果, 你可以以下面这个命令来启动REPL:

命令格式:storm repl

 

【打印本地配置】

命令格式:storm localconfvalue [配置参数关键字]

举例:storm localconfvalue storm.zookeeper.servers

#根据指定参数打印本地配置的值。

 

【打印远程配置】

命令格式:storm remoteconfvalue [配置参数关键字]

举例:storm remoteconfvalue storm.zookeeper.servers

#根据指定参数打印远程配置的值。

 

【执行Shell脚本】

命令格式:storm shell resourcesdir command args

 

【打印CLASSPATH】

命令格式:storm classpath

 

====Storm调优:

--调优对象

当一个topology在storm cluster中运行时,它的并发主要跟3个逻辑对象相关:worker => executor =>task。(=>代表1对N)

(1)Worker

Worker是运行在工作节点上面,被Supervisor守护进程创建的用来干活的JVM进程。

每个Worker对应于一个给定topology的全部执行任务的一个子集。

反过来说,一个Worker里面不会运行属于不同的topology的执行任务。

它可以通过[storm rebalance]命令任意调整。

(2)Executor

可以理解成一个Worker进程中的工作线程。

一个Executor中只能运行隶属于同一个component(spout/bolt)的task。

一个Worker进程中可以有一个或多个Executor线程。在默认情况下,一个Executor运行一个task。

它可以通过[storm rebalance]命令任意调整。

(3)Task

Task则是spout和bolt中具体要干的活了。一个Executor可以负责1个或多个task。

每个component(spout/bolt)的并发度就是这个component对应的task数量。

同时,task也是各个节点之间进行grouping(partition)的单位。无法在运行时调整。

 

--设置方法:

conf.setNumWorkers(workers);                                        //设置worker数量

uilder.setBolt("2", new WordSpliter(),4)                             //设置Executor并发数量

builder.setBolt("2", new WordSpliter(),4).setNumTasks(1); //设置每个线程处理的Task数量

--任务分配:

任务分配是有下面两种情况:

①、task数目比worker多:

例如task是[1 2 3 4],可用的slot(所谓slot就是可用的worker)只有[host1:port1,host2:port1],那么最终是这样分配1:[host1:port1]

2:[host2:port1]

3:[host1:port1]

4:[host2:port1]

②、task数目比worker少:

例如task是[1 2],而worker有[host1:port1,host1:port2,host2:port1,host2:port2],

那么首先会将woker排序,将不同host间隔排列,保证task不会全部分配到同一个机器上,也就是将worker排列成

[host1:port1,host2:port1,host1:port2,host2:port2]

然后分配任务为:

1:[host1:port1]

2:[host2:port1]

 

--简单举例:

通过Config.setNumWorkers(int))来指定一个storm集群中执行topolgy的进程数量,所有的线程将在这些指定的worker进程中运行。

比如说一个topology中要启动300个线程来运行spout/bolt,而指定的worker进程数量是60个。

那么storm将会给每个worker分配5个线程来跑spout/bolt。

如果要对一个topology进行调优,可以调整worker数量和spout/bolt的parallelism(并发度,即executor)数量。

(调整参数之后要记得重新部署topology,后续会为该操作提供一个swapping的功能来减小重新部署的时间)。

 

例如:builder.setBolt("cpp", new CppBolt(), 3).setNumTasks(5).noneGrouping(pre_name); 会创建3个线程,但有内存中会5个CppBolt对象,3个线程调度5个对象。

 

--网上搜罗的一些经验:①、对于worker和task之间的比例,网上也给出了参考,。即1个worker包含10~15个左右。当然这个参考,实际情况还是要根据配置和测试情况。

②、executor数最大不能超过该bolt的task数。

 

--Strom集群命令

12345678910111213141516171819202122232425[root@h2master bin]# storm Commands:         activate         classpath         deactivate         dev-zookeeper         drpc         help 命令帮助         jar   执行上传的jar包         kill   杀死正在执行的topology 后面跟 topology的名称         list   查看运行的所有topology运行情况         localconfvalue         logviewer   启动topology日志         nimbus      启动nimbus         rebalance   shell方式下修改topology运行参数比如worker个数 task个数等         remoteconfvalue         repl         shell         supervisor  启动supervisor         ui              启动topology ui界面         version       Help:          help          help <command>

 

[root@h2master bin]# storm  Commands:          activate          classpath          deactivate          dev-zookeeper          drpc          help 命令帮助          jar   执行上传的jar包          kill   杀死正在执行的topology 后面跟 topology的名称          list   查看运行的所有topology运行情况          localconfvalue          logviewer   启动topology日志          nimbus      启动nimbus          rebalance   shell方式下修改topology运行参数比如worker个数 task个数等          remoteconfvalue          repl          shell          supervisor  启动supervisor          ui              启动topology ui界面          version       Help:           help           help <command>

 

 

--END--


发表评论 共有条评论
用户名: 密码:
验证码: 匿名发表