Storm在zookeeper上的目录结构

eagle

　　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 [this path data])
  (delete-node [this path])
  (create-sequential [this path data])
  ;; if node does not exist, create persistent with this data
  (set-data [this path data])
  (get-data [this path watch?])
  (get-version [this path watch?])
  (get-data-with-version [this path watch?])
  (get-children [this path watch?])
  (mkdirs [this path])
  (close [this])
  (register [this callback])
  (unregister [this id]))
Storm使用Zookeeper的操作
　　

(defprotocol StormClusterState
  (assignments [this callback])
  (assignment-info [this storm-id callback])
  (assignment-info-with-version [this storm-id callback])
  (assignment-version [this storm-id callback])
  (active-storms [this])
  (storm-base [this storm-id callback])
  (get-worker-heartbeat [this storm-id node port])
  (executor-beats [this storm-id executor->node+port])
  (supervisors [this callback])
  (supervisor-info [this supervisor-id]) ;; returns nil if doesn't exist
  (setup-heartbeats! [this storm-id])
  (teardown-heartbeats! [this storm-id])
  (teardown-topology-errors! [this storm-id])
  (heartbeat-storms [this])
  (error-topologies [this])
  (worker-heartbeat! [this storm-id node port info])
  (remove-worker-heartbeat! [this storm-id node port])
  (supervisor-heartbeat! [this supervisor-id info])
  (activate-storm! [this storm-id storm-base])
  (update-storm! [this storm-id new-elems])
  (remove-storm-base! [this storm-id])
  (set-assignment! [this storm-id info])
  (remove-storm! [this storm-id])
  (report-error [this storm-id task-id node port error])
  (errors [this storm-id task-id])
  (disconnect [this]))
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 [master-code-dir node->host executor->node+port executor->start-time-secs])
  |
  |-/storms                 -- 这个目录保存所有正在运行的topology的id
  |   |                        
  |   |
  |   |-/{topology-id}      -- 这个文件保存这个topology的一些信息，包括topology的
  |                            名字，topology开始运行的时间以及这个topology的状态
  |                            操作(active-storms),获得当前路径下活跃的topology数据。保存的内容参考类StormBase
  |                            (storm-base storm-id)得到给定的storm-id下的StormBase数据,具体定义在common.clj中的
  |                            (defrecord StormBase [storm-name launch-time-secs status num-workers component->executors])
  |
  |-/supervisors            -- 这个目录保存所有的supervisor的心跳信息
  |   |                        
  |   |
  |   |-/{supervisor-id}    -- 这个文件保存的是supervisor的心跳信息包括:心跳时间，主
  |                            机名，这个supervisor上worker的端口号运行时间(具体看SupervisorInfo类)
  |                            操作(supervisors)得到所有的supervisors节点
  |                            (supervisor-info supervisor-id)得到给定的supervisor-id对应的SupervisorInfo信息
  |                            具体定义在common.clj中的
  |                            (defrecord SupervisorInfo [time-secs hostname assignment-id used-ports meta scheduler-meta uptime-secs])
  |
  |-/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}