Storm在zookeeper上的目录结构
storm操作zookeeper的主要函数都定义在命名空间backtype.storm.cluster中（即cluster.clj文件中）。 backtype.storm.cluster定义了两个重要protocol：ClusterState和StormClusterState。clojure中的protocol可以看成java中的接口，封装了一组方法。ClusterState协议中封装了一组与zookeeper进行交互的基础函数，如获取子节点函数，获取子节点数据函数等，ClusterState协议定义如下：关于storm操作zookeeper的详细分析请参见博客：storm操作zookeeper源码分析-cluster.clj
Zookeeper的操作
(defprotocol ClusterState
(set-ephemeral-node )
(delete-node )
(create-sequential )
;; if node does not exist, create persistent with this data
(set-data )
(get-data )
(get-version )
(get-data-with-version )
(get-children )
(mkdirs )
(close )
(register )
(unregister ))
Storm使用Zookeeper的操作
(defprotocol StormClusterState
(assignments )
(assignment-info )
(assignment-info-with-version )
(assignment-version )
(active-storms )
(storm-base )
(get-worker-heartbeat )
(executor-beats )
(supervisors )
(supervisor-info ) ;; returns nil if doesn't exist
(setup-heartbeats! )
(teardown-heartbeats! )
(teardown-topology-errors! )
(heartbeat-storms )
(error-topologies )
(worker-heartbeat! )
(remove-worker-heartbeat! )
(supervisor-heartbeat! )
(activate-storm! )
(update-storm! )
(remove-storm-base! )
(set-assignment! )
(remove-storm! )
(report-error )
(errors )
(disconnect ))
Storm中在Zookeeper中存储的目录
(def ASSIGNMENTS-ROOT "assignments")
(def CODE-ROOT "code")
(def STORMS-ROOT "storms")
(def SUPERVISORS-ROOT "supervisors")
(def WORKERBEATS-ROOT "workerbeats")
(def ERRORS-ROOT "errors")
(def ASSIGNMENTS-SUBTREE (str "/" ASSIGNMENTS-ROOT))
(def STORMS-SUBTREE (str "/" STORMS-ROOT))
(def SUPERVISORS-SUBTREE (str "/" SUPERVISORS-ROOT))
(def WORKERBEATS-SUBTREE (str "/" WORKERBEATS-ROOT))
(def ERRORS-SUBTREE (str "/" ERRORS-ROOT)) 从上面来看，在Zookeeper中主要是有如下的五个子目录:
[*]/assignments -> 任务分配信息
[*]/storms -> 正在运行的topology的ID
[*]/supervisors -> 所有的Supervisors的心跳信息
[*]/workerbeats -> 所有的Worker的心跳
[*]/errors -> 产生的出错信息
结构图
/-{storm-zk-root} -- storm在zookeeper上的根目录(默认为/storm)
|
|-/assignments -- topology的任务分配信息
| |
| |-/{topology-id} -- 这个下面保存的是每个topology的assignments信息包括： 对应的
| nimbus上的代码目录,所有task的启动时间,每个task与机器、端口的映射
| 操作为(assignments)来获取所有的assignments的值；
| 以及(assignment-info storm-id)来得到给定的storm-id对应的AssignmentInfo信息
| 在AssignmentInfo中存储的内容有:
| :executor->node+port :executor->start-time-secs :node->host
| 具体定义在common.clj中的
| (defrecord Assignment )
|
|-/storms -- 这个目录保存所有正在运行的topology的id
| |
| |
| |-/{topology-id} -- 这个文件保存这个topology的一些信息，包括topology的
| 名字，topology开始运行的时间以及这个topology的状态
| 操作(active-storms),获得当前路径下活跃的topology数据。保存的内容参考类StormBase
| (storm-base storm-id)得到给定的storm-id下的StormBase数据,具体定义在common.clj中的
| (defrecord StormBase )
|
|-/supervisors -- 这个目录保存所有的supervisor的心跳信息
| |
| |
| |-/{supervisor-id} -- 这个文件保存的是supervisor的心跳信息包括:心跳时间，主
| 机名，这个supervisor上worker的端口号运行时间(具体看SupervisorInfo类)
| 操作(supervisors)得到所有的supervisors节点
| (supervisor-info supervisor-id)得到给定的supervisor-id对应的SupervisorInfo信息
| 具体定义在common.clj中的
| (defrecord SupervisorInfo )
|
|-/workerbeats -- 所有worker的心跳
| |
| |-/{topology-id} -- 这个目录保存这个topology的所有的worker的心跳信息
| |
| |-/{supervisorId-port} -- worker的心跳信息，包括心跳的时
| 间，worker运行时间以及一些统计信息
| 操作(heartbeat-storms)得到所有有心跳数据的topology，
| (get-worker-heartbeat storm-id node port)得到具体一个topology下的某个worker(node:port)的心跳状况，
| (executor-beats storm-id executor->node+port)得到一个executor的心跳状况
|
|-/errors -- 所有产生的error信息
|
|-/{topology-id} -- 这个目录保存这个topology下面的错误信息
| 操作(error-topologies)得到出错的topology
| (errors storm-id component-id)得到给定的storm-id component-id下的出错信息
|-/{component-id}
页:
[1]