黑马程序员技术交流社区

标题: 【上海校区】大数据(十四) - Storm [打印本页]

作者: 不二晨    时间: 2018-9-18 09:20
标题: 【上海校区】大数据(十四) - Storm
【转载】        https://blog.csdn.net/matthewei6/article/details/50570044
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 任务名称






作者: 不二晨    时间: 2018-9-20 17:37
奈斯




欢迎光临 黑马程序员技术交流社区 (http://bbs.itheima.com/) 黑马程序员IT技术论坛 X3.2