http://storm.apache.org/releases/1.0.1/Lifecycle-of-a-topology.html
STORM拓扑的生命周期
本页的内容基于0.7.1代码,后来又很多的变化了。
详细解释了一个拓扑的生命周期: 从提交拓扑,到supervisor启停workers
也解释了Nimbus怎么监视拓扑和拓扑在被killed的时候是怎么关闭的
首先是两个说明
- 真实运行的拓扑和用户声明的拓扑是不一样的,真实的拓扑包含隐含的流和隐含的acker bolt去管理acking 框架(保证了数据的处理),这个隐含的拓扑是通过system-topology!函数实现的
- storm-topology!函数在两个地方被使用
- 当Nimbus为拓扑创建任务的时候
- 在worker中,使得它知道何时路由信息
Starting a topology
- storm jar这个命令用特定的参数去执行你的类,这个命令执行的唯一的特别的事情是设置storm.jar环境变量,为StormSubmitter后来的使用做准备
- 当使用StromSubmitter.submitTopology的时候,StormSubmitter会执行下面的行动
- 首先:如果以前没上传jar会首先上传jar
- 通过Nimbus的Thrift接口Jar上传完成了
- beginFileUpload返回一个Nimbus的inbox的路径
- 15kb 一次上传 通过uploadChunk
- finishFileUpload被调用, 当上传完成的时候
- 这是Thrift的方法实现的Nimbus
- 其次:在Nimbus thrift interface上StormSubmitter调用submitTopology
- 拓扑的配置被JSON序列化了
- 注意Thrift的submitTopology调用需要Nimbus的inbox路径(就是jar上传的路径)
- Nimbus收到了拓扑的提交
- Nimbus规范了拓扑的配置,目的是使得确保每一个任务都有一个相同的序列化注册,这对于序列化工作的正确的执行至关重要。
- Nimbus为拓扑设置静态状态
- Jar包和配置都在本地放着,因为这些文件对于zookeeper来说太大,这些文件被拷贝到{Nimbus local dir}/stormdist/{topology id}
- setup-storm-static写任务-组件映射到ZK
- setup-heartbeats创建ZK目录,在目录中任务可以heartbeat
- Nimbus调用mk-assignment给机器分配任务
- 任务包含:
- master-code-dir:被supervisor使用去下载正确的jars/configs
- task->node+port:一个从任务id到运行它的worker节点的映射
- node->host:从node id到host的映射,使得workers知道和哪个机器去连接实现和其他worker的通信。node id被用来id supervisor,这样多个supervisor可以运行在一个机器上。
- task->start-time-secs:包含一个任务和一个nimbus开始任务的时间戳的映射。这被nimbus使用来监视拓扑。
- 一旦拓扑被指派,它们就初始化为一个不激活的状态。start-storm写数据到Zookeeper使得cluster知道拓扑被激活了,可以从spout发射tuples。
- TODO集群状态图表
- Supervisor在后台运行两个函数
- synchronize-supervisor:这个在zookeeper中的任务改变的时候被调用,平时也是每十秒钟调用一次。
- 对于没有代码的机器,从nimbus给他们下载代码
- 将这个节点应当运行什么:port->LocalAssignment写到文件系统。其中LocalAssignment包含一个拓扑id也包含workers的task id列表
- sync-processes:从本地文件系统读取上一个函数写入的东西,和实际运行着的东西对比。然后必要情况下启停worker进程实现同步。
- worker进程通过mk-worker函数启动
- worker连接另外的worker,并且启动一个线程去 监视变化。如果一个worker得到了重新分配,这个worker就会重连其他的worker的新的状态。
- 监视无论这个拓扑时候是活着的,并且将这个状态存储在storm-active-atom变量中。这个变量被任务用来去决定调用不调用spout的nextTuple方法。
- worker启动多线程处理多任务
- 任务通过mk-task函数设置
- 任务设置路径函数,函数中传入一个流和一个输出tuple并且返回一个任务列表发送给tuple。
- 任务设置spout-specific或者bolt-specific代码
Topology Monitoring
- Nimbus在整个他的生命周期中都监视整个拓扑的状态。
- 调度经常性的任务给timer thread去检查拓扑
- Nimbus的行为表现为一个有限状态机
- 拓扑上的监视时间在每个“nimbus.monitor.freq.secs”被调用,这个通过reassign-transition调用reassign-topology
- reassign-topology调用mk-assignments,这个函数第一次也被用来分配拓扑。mk-assignments也有 能力增加更新拓扑。
- mk-assignments检查心跳分配任务
- 任何的重新分配改变zk中的状态的时候,会触发supervisor去同步,和启停workers。
Killing a topology
- storm kill这个命令将会调用Nimbus Thrift接口去听着一个拓扑
- nimbus接收这个kill
- nimbus执行这个kill过度
- 这个kill过度函数转换拓扑为killed状态,并且转换remove的事件为wait time seconds
- 这个time默认就是timeout但是可以被Kill -w的命令重写
- 导致拓扑被停止,在真实关闭的等待时间给拓扑一个机会在关闭workers之前处理正在处理的事情
- 在Kill 事务中改变状态保证kill协议对Nimbus崩溃 是可以容忍的,一旦启动,如果拓扑的状态是killed,Nimbus调度移除事件运行wait time seconds。
- 移除拓扑,清理任务和zk中的静态信息
- 独立的清理线程运行do-cleanup函数,将会清理本地的heartbeat dir和jars/config
时间: 2024-10-17 10:36:29