A股上市公司传智教育(股票代码 003032)旗下技术交流社区北京昌平校区

 找回密码
 加入黑马

QQ登录

只需一步,快速开始

storm是一个分布式实时计算引擎
storm/Jstorm的安装、配置、启动几乎一模一样
storm是twitter开源的

storm的特点

storm支持热部署,即时上限或下线app
可以在storm上使用各种编程语言如clojure、java、ruby、python等
本地模式:storm有一个本地模式,可以在处理过程中完全模拟storm集群,便于开发和测试。


storm使用场景
    1、流聚合:把两个或多个数据流聚合成一个数据流:基于一些共同的tuple字段
    2、批处理:因为性能或其他原因
    3、BasicBolt:太常见的一种场景,所以storm内置了实现
    4、内存内缓存+fields grouping组合
    5、计算top N
    6、TImecachemapping
    7、分布式DRPC



基本概念
Topology:计算拓扑,即一个应用程序app(通过storm jar发布),因为各个组件间的消息流动形成逻辑上的一个拓扑结构,因此得名。
    TopologyBuilder是拓扑构建器,将spout、bolt等组合起来
spout:消息流的源头,消息生产者。
bolt:消息处理者
Reliability:可靠性,storm保证每个tuple都会被处理。
task:任务,每个spout和bolt都是一个任务,每个任务默认是一个线程。
worker:工作进程,每个工作进程都有多个task
Values:数据容器
Tuple  英[tʌpl]:一个消息传递的基本单元,发送的数据封装到tuple中,实际就是一个value list
        JStorm将流中数据抽象为tuple,一个tuple就是一个值列表value list,list中的每个value都有一个name,并且该value可以是基本类型,字符类型,字节数组等,当然也可以是其他可序列化的类型。
Stream:消息流,源源不断的tuple组成了stream
stream grouping:消息分发策略,一共6种,定义每个bolt接收什么样的消息。
Cofig:设置一些配置信息
StormSubmitter/LocalCluster拓扑提交器

tuple在传输过程中需要序列化和反序列化

http://www.51studyit.com/html/notes/20140329/44.html
spout从外部数据源读取tuple,emit到topology里
spout分可靠的和非可靠的两种,对可靠的,还支持ack和fail方法
Storm Topology是基于Thrift结构, 并且Nimbus是个Thrift server, 所以对于Topology可以用任何语言实现, 最终都是转化为Thrift结构
重要的是, nimbus和supervisor的fail或restart不会影响worker的工作


打比方
Nimbus是老总,下放代码
zookeeper是项目经理,管理集群中的组件,管理任务task,负责nimbus和supervisor协调工作
supervisor是工程主管或技术主管,管理工作进程worker
work是工人,工作进程,他们处理任务task
task即任务,worker进程中每一个spout、bolt、actor的线程都是一个task任务
actor负责跟踪监控任务spout、bolt的执行,他也是任务task
executor是线程,一个task默认由一个executor线程执行,当然一个executor线程里可以处理多个task



storm集群结构
集群由一个主节点和多个子节点(控制节点和工作节点)组成。
1、主节点/控制节点:运行着一个叫做Nimbus的守护进程,负责分配代码、布置任务、故障将侧。
2、子节点/工作节点,运行着一个名为Supervisor的守护进程,负责监听工作,开始并终止工作进程worker
nimubus和storm ui需要在同一台机子上



tuple流的分组机制,即消息分发策略(以下是常用的策略)
    shuffle grouping:随机分组,随机派发stream里的tuple,保证每个bolt收到的tuple数量相同。
    field grouping:按字段分组,如userid,具有相同uiserid的tuple会被分到相同的bolt。(相同userid的tuple会分到相同的bolt中,但是这个bolt中可以有多种tuple)
    all grouping:广播发送,针对一个tuple,所有bolt都会受到(谨慎使用)
    global grouping:全局分组,指全部流都发送到Bolt的同一个任务中,再具体一点,是发送给ID最小的任务。
    non grouping:不分组,不需要关心怎么分组,目前等效于随机分组


storm组件生命周期
Spout方法调用顺序
    declareOutputFields     topology提交过程中会调用spout和bolt的这个方法
    open     类似bolt的prepare方法,而且参数也类似,可以处理配置信息,做准备工作
    activate
    nextTuple    循环调用
    deactivate
Bolt方法调用顺序
    declareOutputFields
    prepare
    execute    循环调用


storm可靠性
storm有默认的配置文件,在storm jar包里 storm.yaml
storm有一种机制可以保证从spout发出的每个tuple都会被完全处理。
什么样的消息被认为完整处理了:
    1、tuple tree不再生长
    2、树中的任何消息都标识为"已处理"



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在哪些路径下定位本地库.


命令

发布启动任务
    storm jar jar包名 包含main方法的类
    storm jar jar包名 包含main方法的类  拓扑名
查看任务
    storm list
停止任务
    storm kill 任务名称




1 个回复

倒序浏览
奈斯,加油加油
回复 使用道具 举报
您需要登录后才可以回帖 登录 | 加入黑马