Storm系列(四)Topology校验过程

时间：2015-09-08 18:28:47 阅读：332 评论：0 收藏：0 [点我收藏+]

标签：

功能：提交一个新的Topology,并为Topology创建storm-id（topology-id）,校验其结构，设置必要的元数据，最后为Topology分配任务.

实现源码：

            1            (^void submitTopology       
         2                    [this ^String storm-name ^String uploadedJarLocation ^String serializedConf ^StormTopology topology]       
         3                    (.submitTopologyWithOpts this storm-name uploadedJarLocation serializedConf topology       
         4                                             (SubmitOptions. TopologyInitialStatus/ACTIVE)))       
 

从以上源码中看出submitTopology内部是对submitTopologyWithOpts方法的调用。

submitTopologyWithOpts函数原型如下:

            1            ^void submitTopologyWithOpts       
         2                    [this ^String storm-name ^String uploadedJarLocation ^String serializedConf ^StormTopology topology       
         3                     ^SubmitOptions submitOptions]       
 

在submitTopologyWithOpts中主要做了以下几件事情：

校验submitOptions参数不能为空。
检查storm-name中是否包含非法字符。
校验storm-name与正在运行的Topology是否有重名，重名将造成冲突。
将nimbus（nimbus-data类型）中的submitted-count已提交Topology计数字段加1。
为所提交的Topology创建唯一的storm-id(topology-id),格式：<storm-name>-<submitted-count>-<当前时间>
通过normalize-conf获取提交的Topology的Storm配置,首先将参数serializedConf进行反序列化，然后加入storm-name,storm-id等。
将Storm默认的配置(conf)与第六步得到的Storm配置进行合并，合并原则为两份配置中重复的配置项以第六步中的配置为准。
调用normalize-topology计算提交的Topology中每个组件并行度及更新TOPOLOGY_TASKS配置项.
获取nimbus（nimbus-data类型）中storm-cluster-state对象。
调用System-topology!方法对Topology结构进行校验。
获取nimbus中的submit-lock锁。
调用setup-storm-code为Topology创建对应的本地文件夹、复杂jar并写入序列化后的Storm配置项和Topology信息.
调用setup-hearbeats!为Topology在Zookeeper中创建心跳路径，/storm/workerbeats/topology-id.
定义一个从thrift-status到keyword-status的哈希表，该哈希表用来将传入的submitOptions中的thrift-status转化为对应的keyword-status.
调用start-storm设置stormBase,它在Zookeeper中路径是/storm/storms/<topology-id>,stormBase的信息将做为该路径所对应的存储值。
调用mk-assignments为所提交的Topology分配资源.

normalize-topology

实现源码:

            1            (defn normalize-topology [storm-conf ^StormTopology topology]       
         2              (let [ret (.deepCopy topology)]       
         3                (doseq [[_ component] (all-components ret)]       
         4                  (.set_json_conf       
         5                    (.get_common component)       
         6                    (->> {TOPOLOGY-TASKS (component-parallelism storm-conf component)}       
         7                         (merge (component-conf component))       
         8                         to-json )))       
         9            ret ))       
 

实现说明:

调用deepCopy对topology进行深度拷贝，赋值给ret.
遍历topology（ret）所有组件，调用component-parallelism更新组件配置中的TOPOLOGY_TASKS信息。

component-parallelism实现源码(计算组件并行度):

            1            (defn- component-parallelism [storm-conf component]       
         2              (let [storm-conf (merge storm-conf (component-conf component))       
         3                    num-tasks (or (storm-conf TOPOLOGY-TASKS) (num-start-executors component))       
         4                    max-parallelism (storm-conf TOPOLOGY-MAX-TASK-PARALLELISM)       
         5                    ]       
         6                (if max-parallelism       
         7                  (min max-parallelism num-tasks)       
         8                  num-tasks)))       
 

实现说明：

将Topology配置信息与组件(component)配置信息进行合并，两者存在重复的配置项时以组件的配置项为准。
计算组件并行度(num-tasks)，若果配置storm-conf中配置了TOPOLOGY-TASKS信息，就以该配置值做为组件的并行度，否则通过调用num-start-executors获取用户对组件设置的并行度做为num-tasks.
获取storm-conf配置中TOPOLOGY-MAX-TASK-PARALLELISM配置项的值。
返回TOPOLOGY-MAX-TASK-PARALLELISM与num-tasks较小的值做为组件的并行度。

            1            TopologyBuilder builder = new TopologyBuilder();       
         2            // 4对应对用用户设置的组件并行度，10对应TOPOLOGY-TASK配置项的值       
         3            builder.setBolt("transfer", new TransferBolt(), 4).shuffleGrouping("random").setNumTasks(6); Config conf = new Config();       
         4            // 8对应 TOPOLOGY-MAX-TASK-PARALLELISM配置项的值       
         5            Conf.setMaxTaskParallelism（8）;       
 

system-topology!

功能：

验证用户提交的Topology,同时为提交的topology添加一些系统组件和流。

实现源码：

            1            (defn system-topology! [storm-conf ^StormTopology topology]       
         2              (validate-basic! topology)       
         3              (let [ret (.deepCopy topology)]       
         4                (add-acker! storm-conf ret)       
         5                (add-metric-components! storm-conf ret)           
         6                (add-system-components! storm-conf ret)       
         7                (add-metric-streams! ret)       
         8                (add-system-streams! ret)       
         9                (validate-structure! ret)       
         10                ret       
         11            ))       
 

实现说明：

使用validate-basic!校验所提交的Topology.
主要用于确保topology中的组件id不重复而且不是系统id，以及确保每个组件的TOPOLOGY-TASKS配置项大于0时，组件的并行度设置也一定大于0.
调用deepCopy对topology进行深度拷贝，赋值给ret.
为Topology添加acker-bolt.
用于追踪发送出去的消息是否被成功处理。
使用add-metric-components为Topology添加metric-bolt.
为Topology添加system-bolt.
System-bolt没有输入流只有输出流分别为：SYSTEM-TICK-STREAM-ID，声明字段是[“rate_secs”],非直接模式；另一个为METRICS-TICK-STREAM-ID,声明字段为[“interval”]非直接模式，并行度为0.
为Topology中的所有组件添加统计流。
Stream-id为METRICS-STREAM-ID,声明字段为[“task-info”,”data-points”],非直接流模式.
为Topology中的所有组件添加系统流。
stream-id为SYSTEM-STREAM-ID,声明字段为[“event”],非直接流模式.
使用validate-structure!检验以上步骤所组合后的Topology.

验证过程：
获取Topology中所有组件和组件的输入(包括component-id、stream-id、Grouping),对输入组件依次判断输入组件ID(component-id)是否在该Topology中，不存在则抛出异常，存在则再判断该组件的流类型是否为所对应的stream-id，若不存在则抛出异常，存在则继续检查该流的分组方式（Grouping）是否与能对应，所有组件检查完毕后没有异常抛出表示该Topology有效.

Storm系列(四)Topology校验过程

标签：

原文地址：http://www.cnblogs.com/jianyuan/p/4792443.html

踩

(0)

评论一句话评论（0）

分享档案

更多>

2021年07月29日 (22)
2021年07月28日 (40)
2021年07月27日 (32)
2021年07月26日 (79)
2021年07月23日 (29)
2021年07月22日 (30)
2021年07月21日 (42)
2021年07月20日 (16)
2021年07月19日 (90)
2021年07月16日 (35)

周排行

1	(^void submitTopology
2	[this ^String storm-name ^String uploadedJarLocation ^String serializedConf ^StormTopology topology]
3	(.submitTopologyWithOpts this storm-name uploadedJarLocation serializedConf topology
4	(SubmitOptions. TopologyInitialStatus/ACTIVE)))

1	^void submitTopologyWithOpts
2	[this ^String storm-name ^String uploadedJarLocation ^String serializedConf ^StormTopology topology
3	^SubmitOptions submitOptions]

1	(defn normalize-topology [storm-conf ^StormTopology topology]
2	(let [ret (.deepCopy topology)]
3	(doseq [[_ component] (all-components ret)]
4	(.set_json_conf
5	(.get_common component)
6	(->> {TOPOLOGY-TASKS (component-parallelism storm-conf component)}
7	(merge (component-conf component))
8	to-json )))
9	ret ))

1	(defn- component-parallelism [storm-conf component]
2	(let [storm-conf (merge storm-conf (component-conf component))
3	num-tasks (or (storm-conf TOPOLOGY-TASKS) (num-start-executors component))
4	max-parallelism (storm-conf TOPOLOGY-MAX-TASK-PARALLELISM)
5	]
6	(if max-parallelism
7	(min max-parallelism num-tasks)
8	num-tasks)))

1	TopologyBuilder builder = new TopologyBuilder();
2	// 4对应对用用户设置的组件并行度，10对应TOPOLOGY-TASK配置项的值
3	builder.setBolt("transfer", new TransferBolt(), 4).shuffleGrouping("random").setNumTasks(6); Config conf = new Config();
4	// 8对应 TOPOLOGY-MAX-TASK-PARALLELISM配置项的值
5	Conf.setMaxTaskParallelism（8）;

1	(defn system-topology! [storm-conf ^StormTopology topology]
2	(validate-basic! topology)
3	(let [ret (.deepCopy topology)]
4	(add-acker! storm-conf ret)
5	(add-metric-components! storm-conf ret)
6	(add-system-components! storm-conf ret)
7	(add-metric-streams! ret)
8	(add-system-streams! ret)
9	(validate-structure! ret)
10	ret
11	))