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 任务名称
|
|