Storm源码阅读总结 -- Client Nimbus Supervisor

2017/02/22 10:43
阅读数 131

Client 客户端提交作业 NimbusClient: RPC 客户端, 向RPC服务端即Nimbus Server发起RPC调用.

App通过StormSubmitter提交计算拓扑作业submitTopology: 首先提交jar包, 会向Nimbus服务器发起beginFileUpload, 申请到要上传的路径后, 调用uploadChunk开始将jar包上传到nimbus服务器 文件传输完毕, 调用finishFileUpload结束上传jar包过程. 最后会调用NimbusClient的submitTopology

Nimbus启动初始化 Nimbus启动时, 创建匿名的INimbus内部类. 初始时会根据系统中的supervisors创建可用的WorkSlot空闲槽位. --> allSlotsAvailableForScheduling

在一个机器上一个supervisor可以配置多个端口, 每个端口都会对应一个工作者线程, 即创建对应数量的WorkSlot. Nimbus要分配任务给Supervisor上的Worker进行工作, 而每个Supervisor会有多个worker.

Thrift RPC步骤 编写thrift文件 自动生成接口相关的类, 比如Iface, Processor, Client内部类 自定义Handler实现Iface业务逻辑, 对应thrift文件中的方法 服务端使用ServerTransport和Handler创建的Processor创建出Server并启动服务器 客户端使用Transport连接服务端, 使用Protocol协议构造出Client代理类, 然后调用接口的方法, 完成RPC调用

Nimbus Thrift RPC Server 启动Nimbus会启动它的thrift服务. Nimbus会作为Thrift RPC协议的服务端, 处理客户端发起的RPC调用请求. nimbus的thrift服务的实现类定义在storm.thrift文件中, 对应的实现方法是service-handler[服务处理器].

Nimbus与ZooKeeper Nimbus启动时会创建调度器, 它为集群中的需要调度的计算拓扑分配任务. 新的任务可以通过cluster.getAssignments()获取.

nimbus在启动时会连接ZooKeeper, 并在zookeeper中创建节点 以便在作业运行过程中将Storm的所有的状态信息都是保存在Zookeeper里面.

nimbus通过在zookeeper上面写状态信息来分配任务, supervisor,task通过从zookeeper中读状态来领取任务, 同时supervisor, task也会定义发送心跳信息到zookeeper, 使得nimbus可以监控整个storm集群的状态, 从而可以重启一些挂掉的task

注意Nimbus和Supervisor之间没有直接交互, 状态都是保存在Zookeeper上

问题1: storm的jar包是上传到zookeeper上,还是nimbus, supervisor上? 答: jar包上传到nimbus上, 在zookeeper的assignments中只是记录了topology作业在nimbus上的代码目录.

watcher和callback storm使用curator framework创建districtbuted-cluster-state[分布式的集群状态]. 创建客户端时指定watcher函数, 当节点状态发生改变时, 会触发curator上注册的监听器, 回调watcher方法.

注册的watcher使用callbacks. ClusterState自定义的register实现会将回调实现传入, 最终触发回调的调用. callback和watcher都是function, 都通过回调的方式.

Storm的集群状态StormClusterState包括任务的分配assignments, supervisors, workers, hearbeat, errors等信息

Topology执行流程 定义在nimbus的service-handler的submitTopology里, 步骤包括: 上传topology代码 运行topology前的校验 在nimbus上建立topology的本地目录: setup-storm-code 建立Zookeeper heartbeats: setup-heartbeats! 启动storm: start-storm, 在zk上写入StormBase信息,会记录component和任务并行度的关系. 用于后面任务的分配. 分配任务: mk-assignments

Topology Supervisor Worker Executor Task 1. topology 物理上由一个nimbus机器和多台supervisor机器组成

supervisor 可以配置多个slots. 每个slots占用一个端口, 配置在storm.yaml的supervisor.slots.ports 一个worker对应supervisor的一个端口. 因此一个supervisor配置了几个slots, 就对应有几个worker. topology 逻辑上由 spout 和 bolt 组成. spout和bolt统称为component spout和bolt都可以配置并行度, 这个并行度的数量就是component的executors的数量. component如果没有设置task的数量, 默认一个executor运行一个task. 如果设置了task的数量(task数量要大于并行度的值), 则一个executor可以运行多个task. 当然因为executor是针对具体component的(即指定的spout或bolt), 所以executor里运行的多个task都是同一种component.

executors 是topology中所有component每个task起始和结束编号的序列: ([start-task end-task]), 但是没有对应的component信息. executors->component 是一个Map: {[start-task end-task] component-id}, 给上面的executors添加了component-id信息. topology->executors 也是一个Map: {storm-id [start-task end-task]}, 把executors作为topology-id的value 假设builder.setSpout("spout", new RandomSentenceSpout(), 5).setNumTasks(10); 则表示RandomSentenceSpout这个component有5个executors, 设置了10个任务tasks. 那么这个component的executors = ([1 2] [3 4] [5 6] [7 8] [9 10]). 对应的executors->component = {[1 2] "spout" [3 4] "spout" [5 6] "spout" [7 8] "spout" [9 10] "spout"} 注意: executors和executors->component的数据应该是包括了topology所有的component! 这里因为只列出一个component,所以只给出部分数据.

executor是属于逻辑上的, 而任务是要运行在物理机器上的. 所以就涉及到逻辑上的executor运行在物理上的supervisor的问题. 而我们知道supervisor通过slots端口配置可以分成多个worker. worker由node-id+port组成. node-id实际上就是supervisor-id. 因此一个worker可以运行多个executor. 每个executor是运行在node-id + port上的. executor->node+port 的结构是: {executor [node port]}. 表示每个executor运行在哪个supervisor的哪个端口上. 通过这种方式可以将同一个component的多个executor分布在多个机器上执行 topology->executor->node+port 的结构: {topology-id -> {executor [node port]}} 任务的分配 产生executor->node+port关系, 将executor分配到哪个node的哪个slot上

mk-assignments的第一步的流程: 准备阶段读取所有active topology的信息

  1. storm-cluster-state.active-storms 获取集群中所有活动的topology-id. 对每个topology-id做如下处理:
  2. read-topology-details 根据topology-id读取每个topology的详细信息, 并组合成{topology-id TopologyDetails} 2.1 storm-cluster-state.storm-base 读取/storm/storms/topology-id节点的数据为StormBase对象,获得保存在其中的numWorkers 2.2 读取storm-conf和storm-topology形成topology-conf和topology对象 2.3 executor->component 的映射关系: {ExecutorDetails component-id} 2.3.1 compute-executor->component 返回: {[start-task end-task] component-id} 2.3.1.1 component->executors 是component-id和executors并行度的数量的映射 2.3.1.2 storm-task-info 给每个executor添加编号 2.3.1.3 compute-executors 列出所有executor的[start-task end-task]序列 2.3.1.4 经过join等一些列操作, 形成最后的{[start-task end-task] component-id} 2.3.2 将上面的[start-task end-task]封装成ExecutorDetails, 形成最终的{ExecutorDetails component-id} 2.4 将上面的storm-id, topology-conf, topology, numWorkers, executor->component构造成TopologyDetails
  3. 将所有的{topology-id TopologyDetails} 封装成Topologies.

分配新任务 计算拓扑和compoent的并行度: topology->executors和topology->alive-executors 找出Supervisor中dead的WorkerSlot: supervisor->dead-ports, 用于得到Supervisor的信息SupervisorDetail 为alive-executors生成SchedulerAssignment: topology->scheduler-assignment 用supervisors信息和topology->scheduler-assignment会生成Cluster Nimbus的调度器开始调度拓扑作业, 使用集群信息topologies和cluster开始调度作业 分配的任务会写到zk的assignments节点中, 下次调度时会根据已经分配的任务找出需要新分配的任务

计算拓扑和集群信息 Topologies包含当前集群里面运行的所有Topology的信息:StormTopology对象,配置信息, 以及从task到组件(bolt, spout)id的映射信息(executor->component-id)。

Cluster对象则包含了当前集群的所有状态信息:所有的supervisor信息, 系统所有Topology的task分配信息(topology->scheduler-assignment, executor->slot),

任务调度 nimbus的任务分配算法: 在slot充沛的情况下,能够保证所有topology的task被均匀的分配到整个机器的所有机器上 在slot不足的情况下,它会把topology的所有的task分配到仅有的slot上去,这时候其实不是理想状态,所以。。 在nimbus发现有多余slot的时候,它会重新分配topology的task分配到空余的slot上去以达到理想状态。 在没有slot的时候,它什么也不做

Nimbus的边城世界 nimbus中最重要的方法是mk-assignments分配任务. 它的调用有几个地方:

service-handler的定时线程schedule-recurring会定时调用. service-handler的匿名内部类Nimbus$Iface的submitTopology第一次提交拓扑作业时 do-rebalance进行负载均衡时, 是由state-transitions来判定nimbus的状态达到rebalance状态时调用的. 第一次提交的topology作业, 就会立刻调用mk-assignments由nimbus分配任务. 但是作业一提交并不一定能够分配到任务. 比如集群的计算资源非常紧张没有可用slot的情况下. 因此需要一个定时器定时调用mk-assignments方法, 在计算资源可用的情况下为没有分配的topology分配任务. 也有一种可能是初次提交作业后, nimbus只为这个topology分配了一部分任务, 因此在下一轮回时还要继续为这个作业分配任务. 这种情况常见于: 一个很长的topology流计算作业, 前面的blot任务还没完成时, 如果为后面的bolt分配任务, 后面的blot任务就一直占用这计算资源, 与其这样, 还不如集中精力把计算资源都分配给前面的blot task. 或者是集群的资源比较紧张, 只够分配topology的一部分任务, 剩余的任务由于没有可用的空闲槽位需要等到下次才能申请到资源.

mk-assignments中最重要的方法当属compute-new-topology->executor->node+port 计算新的topology到executor, 再到executor所属的supervisor node+port的映射关系.

用户编写的topology作业是由自定义的spout和blot组成, 它们统称为component. 可以为component定义并行度, 这就是topology的executors. executors实际上是自定义任务的封装. 集群的supervisor可配置多个port, 对应的是集群中的计算资源WorkerSlot. 任务是要运行在WorkerSlot上的. 所以分配任务就是要将executor分配到指定的WorkerSlot上.

由于上面集中作业的任务分配的不确定性, 为topology分配完任务后, 需要记录已经分配了哪些任务, 这样下次的任务分配就不会为已经分配的任务再次申请计算资源了. nimbus将任务的分配写入到zk的assignments中.

任务分配后要进行调度才算真正的执行. 调度器的工作调用cluster.assign传入 作业topology, 计算资源worker-slot和表示任务的executors. 通过topology得到作业的SchedulerAssignment, 将slot和executors传给SchedulerAssignment的assign. 虽然SchedulerAssignmentImpl只是在内存中记录了ExecutorDetails和WorkerSlot的映射关系. 但是这对于任务的运行而言已经足够了, 因为任务编号记录在ExecutorDetails中, 任务执行的节点记录在WorkerSlot里. 接下来任务就可以真正跑在集群的计算资源里了.

Supervisor 物理上的Supervisor配置了多个端口, 对于集群而言工作节点就是计算资源. 一个端口是一个WorkerSlot. Nimbus负责任务的调度分配, 将任务分配信息Assignment写到ZooKeeper中. Supervisor会负责读取ZK中的任务分配信息.

mk-synchronize-supervisor和synchronize-processes是zk的assignment节点发生变化触发执行的回调函数. 当有任务写到ZK中, Supervisor会调用回调函数同步topology代码到本地, 由于Supervisor负责启动Worker进程,也会同步Worker信息. 同步topology和worker信息的事件会放到一个队列线程里异步地执行.

同步supervisor会从nimbus下载topology代码到本地目录, 并持久化到LocalState状态信息里. 同步worker会找出新分配的worker-id, 负责启动Worker进程.

Nimbus是集群的总管, 只有一台. 而Supervisor监督者有多个, 对于Master-Slave的架构, Slave要定期发送心跳信息给Master. 一个监督者也有多个Worker, 所以Worker也要发送心跳信息给Supervisor. Supervisor和Worker的心跳信息都保存在ZK节点上. 但是注意supervisor从nimbus下载的topology信息和已经处理完成的worker信息是保存在supervisor的本地目录中.

展开阅读全文
打赏
0
0 收藏
分享
加载中
更多评论
打赏
0 评论
0 收藏
0
分享
返回顶部
顶部