图文精华

About云-梭伦科技»专题 › 技术学习(版主发帖区) › 大数据学习 › Storm › Storm项目：流数据监控五Zookeeper统一配置
分享

Storm项目：流数据监控五Zookeeper统一配置

tntzbzc 发表于 2014-11-16 13:54:24 [显示全部楼层] 回帖奖励 阅读模式 关闭右栏 2 36461
About云VIP帮助找工作和提升
问题导读
1.想要实现集群配置文件的统一配置,什么方案比较合适？
2.如何实现zookeeper统一配置









1 文档说明     
很早以前就有朋友提出过storm的拓扑任务会出现supervisor各节点配置麻烦的问题，基于此，我很早就考虑如何解决这个问题，直到现在才把这个问题解决（工作小忙）。
    使用zookeeper进行设计storm拓扑的配置系统，设计思路大家可以参考参考，代码部分就有点次了（时间太匆忙了，白天上班，大半夜写代码，写文章不容易啊），大家能看得下去就看吧，不过优化优化改改还是能用的。

2 Zookeeper统一配置

2.1 设计统一配置因素在storm的拓扑任务中
采用常规的节点配置从配置文件（conf）中直接读取，会出现以下几个问题：
(1) 节点太多，配置文件需要配置的节点太多，太繁琐；
(2) 配置需要修改时，很难保证各个节点的配置文件统一；
(3) 无法定位各个功能节点位置，每个节点需要所有配置文件，导致每个节点配置臃肿；
(4) 配置文件太分散，动态更新不现实。
以上几个主要原因导致了，采用传统的直接读取配置文件获取配置信息的方式不适合类似这种集群应用，必然需要设计一个能够进行统一配置的组件或者是系统。

2.2 配置系统技术选型    

想要实现集群配置文件的统一配置，必然不能将配置文件放在各个分散的节点上，考虑的思路是放在一个节点上，然后让该节点读取，其他节点从该节点获取，这样就保证了配置文件统一性（只此一份）。可以考虑以下几种设计思路：

(1) 在storm的拓扑主类（Topology）中使用Config conf = new Config();进行参数传递，及在主类中读取配置文件中的所有配置，然后通过Config类传递给相应节点。这种方式，笔者曾经尝试过，确实能解决配置文件统一问题，只需nimbus节点读取配置即可，但是，这种方式在配置文件很多时，会导致主类Topology相当臃肿，相当麻烦，并且相应节点获取参数时也相当麻烦。

(2) 考虑使用mysql。在nimbus中，读取配置文件，然后将配置信息按一定规则存放到一个所有节点都能访问的mysql服务器中。该mysql服务器专门用来保存集群的配置文件，所有节点都上该mysql服务器中读取配置信息，这种方式能够解决配置统一的问题，并且也不少集群应用是这么干的。但这种方式有不少的缺点，首先是需要部署一个专门的mysql服务器，这是额外的代价，如果部署到其中一个节点中倒是可以，但也会影响性能，并且mysql并没有对配置信息变更有效的监控措施，并且存在单点问题，安全性，实用性都不太高。

(3) 还有就是使用zookeeper来实现了。Zookeeper的其中有个功能有点类似向zookeeper集群中提供公共文件夹（官方语言称其为“节点”）。能够访问zk服务器的服务器都能够对其公共节点进行访问。利用该机制，我们就可以使用zk设计统一配置系统了。Nimbus以客户端的身份向zk服务器申请公共节点，并在公共节点上存储配置信息，其他节点需要获取配置文件时，以客户端的身份从zk服务器中获取公共节点的相关值，这样就实现了统一配置。实用zk进行统一配置有如下好处：storm集群中已经有zk，不需要额外部署；zk无单点故障问题，配置信息安全性高；zk对节点具有watch功能，节点变更能够进行提醒并可以自定义相关操作。所以就目前来说，使用zk进行系统统一配置是比较合理的。


2.3 统一配置设计方案


图2.3 Zookeeper统一配置架构图


(1) 如图2.3所示，nimbus读取Topology.conf文件，将除nimbus节点配置参数以外的配置信息放入zookeeper服务器（zk集群）的最底层公共节点上（创建一定规则的节点树，放在最底层的子节点上）；

(2) zookeeper集群维护这些配置节点，进行永久保持（除非主动删除），并执行watcher，监视配置节点的变更，随时通知supervisor配置变化；

(3) supervisor第一次从zk集群中读取配置信息，然后开始检测zk的watcher信号，当检测到wather信号（配置变更），重新获取配置信息，从而不但实现了统一配置，还能实现配置的在线动态更新（这点非常重要）；

(4) 整个zk中配置节点目录层次为第一层是总配置根节点/TopologyConf；第二层则是根据拓扑提交的任务名进行创建，使用任务民进行配置节点创建不会导致有重复配置出现，因为一个storm集群中，拓扑的任务名是唯一的；底层节点为各个存储配置的名称，值为配置参数。
     整个架构围绕zookeeper而存在，而zookeeper也是storm集群的依赖集群，需要额外配置，只需跟storm共享zookeeper集群即可，并且不需要担心配置系统会出现故障，zookeeper集群不存在系统故障。
这个架构不单纯可以在storm的应用集群中使用，其他类似的集群应用配置都可以进行类似设计，能够进行统一配置，无单点故障，实现动态更新。


2.4 代码参考

PS：这部分代码由于时间关系，写的比较糙，只是实现了配置信息的存取，没有实现wather功能，并且相对于之前版本的工程代码，变化较大，舍弃了所有可观性不强的xml方式配置，使用标准方式配置格式，并把所有配置相关的类整合在一个包里（storm.conf），供主类及其他bolt调用。
2.4.1 storm.conf.OptZookeeper.java

  1. 该类主要定义了一些zookeeper的基本操作，作为一个基本类供其他conf类调用，其中主要的一些方法定义如下：

  2.      private static final int SESSION_TIMEOUT = 3000;

  3.     private ZooKeeper zookeeper0;

  4.     // 以下为各种zk操作方法

  5.     private Watcher watcher = new Watcher() {

  6.         public void process(WatchedEvent event) {

  7.             System.out.println("process: " + event.getType());

  8.         }

  9.     };

  10.     // 链接zk,参数为连接zookeeper的server地址及端口，可设置多个以“,”隔开

  11.     // 如zkServer=192.168.2.240:2181,192.168.2.241:2181

  12.     public ZooKeeper connect(String zkServer) throws IOException {

  13.         zookeeper0 = new ZooKeeper(zkServer, SESSION_TIMEOUT, watcher);

  14.         return zookeeper0;

  15.     }

  16.     // 关闭zk

  17.     public void close(ZooKeeper zookeeper) throws InterruptedException {

  18.         zookeeper.close();

  19.     }

  20.     // 创建节点

  21.     /**

  22.      * 创建一个znode 1.CreateMode 取值 PERSISTENT：持久化，这个目录节点存储的数据不会丢失

  23.      * PERSISTENT_SEQUENTIAL

  24.      * ：顺序自动编号的目录节点，这种目录节点会根据当前已近存在的节点数自动加1，然后返回给客户端已经成功创建的目录节点名；

  25.      * EPHEMERAL：临时目录节点，一旦创建这个节点的客户端与服务器端口也就是session过期超时，这种节点会被自动删除

  26.      * EPHEMERAL_SEQUENTIAL：临时自动编号节点

  27.      * @throws InterruptedException

  28.      * @throws KeeperException*/

  29.     // 若节点存在则只设置值，若不存在则创建，若出错返回false

  30.     public boolean createNode(ZooKeeper zookeeper, String node, String value) {

  31.         try {

  32.             if (zookeeper.exists(node, null) != null) {

  33.                 zookeeper.setData(node, value.getBytes(), -1);

  34.             } else {

  35.                 zookeeper.create(node, value.getBytes(), Ids.OPEN_ACL_UNSAFE,

  36.                         CreateMode.PERSISTENT);

  37.             }

  38.         } catch (Exception e) {

  39.             System.out.println(e.getMessage());

  40.             Assert.fail();

  41.             return false;

  42.         }

  43.         return true;

  44.     }

  45.     // 删除节点，忽略版本,若不存在则抛错

  46.     public boolean deleteNode(ZooKeeper zookeeper, String node) {

  47.         try {

  48.             if (zookeeper.exists(node, null) != null) {

  49.                 zookeeper.delete(node, -1);

  50.                 return true;

  51.             } else {

  52.                 System.out.println(node + " exists now!");

  53.                 return false;

  54.             }

  55.         } catch (Exception e) {

  56.             System.out.println(e.getMessage());

  57.             Assert.fail();

  58.             return false;

  59.         }

  60.     }

  61.     // 获取数据，若节点不存在抛错

  62.     public String getData(ZooKeeper zookeeper, String node) {

  63.         String result = null;

  64.         try {

  65.             if (zookeeper.exists(node, null) != null) {

  66.                 byte[] bytes = zookeeper.getData(node, null, null);

  67.                 result = new String(bytes);

  68.             } else {

  69.                 System.out.println(node + " exists now!");

  70.             }

  71.             // 类型转换

  72.         } catch (Exception e) {

  73.             // TODO Auto-generated catch block

  74.             System.out.println(e.getMessage());

  75.             Assert.fail();

  76.         }

  77.         return result;

  78.     }

  79.     // 获取数据，设置watch,只要节点改变（节点的值改变）则触发watcher，执行watcher中函数 

  80.     public String getDataWatch(ZooKeeper zookeeper, String node) {

  81.         String result = null;

  82.         try {

  83.             if (zookeeper.exists(node, null) != null) {

  84.                 byte[] bytes = zookeeper.getData(node, new Watcher() {

  85.                     public void process(WatchedEvent event) {

  86.                         System.out.println("testExistWatch2 Watch : {}"

  87.                                 + event.getType());

  88.                     }

  89.                 }, null);

  90.                 result = new String(bytes);

  91.             } else {

  92.                 result = null;

  93.             }

  94.         } catch (Exception e) {

  95.             // TODO Auto-generated catch block

  96.             System.out.println(e.getMessage());

  97.             Assert.fail();

  98.         }

  99.         return result;

  100.     }

  101.     // 设置对应目录下的数据，-1表示忽略版本

  102.     // 该设置能改变原始值

  103.     public boolean setData(ZooKeeper zookeeper, String node, String value) {

  104.         try {

  105.             if (zookeeper.exists(node, null) != null) {

  106.                 zookeeper.setData(node, value.getBytes(), -1);

  107.                 return true;

  108.             } else {

  109.                 return false;

  110.             }

  111.         } catch (Exception e) {

  112.             System.out.println(e.getMessage());

  113.             Assert.fail();

  114.             return false;

  115.         }

  116.     }

  117.     // 获取子节点,返回一个list

  118.     public List getNodeChild(ZooKeeper zookeeper, String node) {

  119.         try {

  120.             if (zookeeper.exists(node, null) != null) {

  121.                 List list = zookeeper.getChildren(node, true);

  122.                 return list;

  123.             } else {

  124.                 return null;

  125.             }

  126.         } catch (Exception e) {

  127.             System.out.println(e.getMessage());

  128.             Assert.fail();

  129.             return null;

  130.         }

  131.     }
复制代码



2.4.2 storm.conf.GetTopConf.java

该类主要是提供给主类相关的配置信息，由于主类是在nimbus上执行，所以其所需配置文件则直接获取，不经过zookeeper配置系统。

  1.     public void getTopConf(String configFile) throws Exception {

  2.         Properties pros = new Properties();

  3.         File config = new File(configFile);

  4.         pros.load(new FileInputStream(config));

  5.         // 获取规则列表路径

  6.         this.TopologyList = pros.getProperty("topologyList.path",

  7.                 "TopologyList.rule");

  8.         // 获取worker数值

  9.         this.WorkerNum = pros.getProperty("worker.num", "10");

  10.         if (isNum.isNumeric(WorkerNum)) {

  11.             this.MaxWorker = Integer.parseInt(WorkerNum);

  12.         }

  13.         if (TopologyList != null) {

  14.             FileInputStream fis = new FileInputStream(TopologyList);

  15.             InputStreamReader isr = new InputStreamReader(fis, "UTF-8");

  16.             BufferedReader br = new BufferedReader(isr);

  17.             String line = "";

  18.             // 计算规则列表有效节点数

  19.             while ((line = br.readLine()) != null) {

  20.                 if (line.length() != 0) {

  21.                     this.Topology[this.Top_num] = line;

  22.                     this.Top_num++;

  23.                 }

  24.             }

  25.         }

  26.     }
复制代码

  1.   private static final int SESSION_TIMEOUT = 3000;

  2.     private ZooKeeper zookeeper;

  3.     private OptZookeeper optzk = new OptZookeeper();

  4.     public void putZkConf(String configFile, String taskName) throws Exception {

  5.         Properties pros = new Properties();

  6.         File config = new File(configFile);

  7.         pros.load(new FileInputStream(config));

  8.         // 从nimbus域中获取配置文件zk服务器地址并连接zk

  9.         String nimbusZkConect = pros.getProperty("nimbus.zkConnect",

  10.                 "localhost:2181");

  11.         zookeeper = optzk.connect(nimbusZkConect);

  12.         // 创建配置根目录

  13.         String ConfRoot = "TopologyConf";

  14.         String StrTmp = "/";

  15.         String node1 = StrTmp + ConfRoot;

  16.         String node2 = StrTmp + ConfRoot + StrTmp + taskName;

  17.         // 创建一级根目录"/TopologyConf"

  18.         if (zookeeper.exists(node1, null) == null) {

  19.             optzk.createNode(zookeeper, StrTmp + ConfRoot, "TopologyConf");

  20.         }

  21.         // 创建二级根目录"/TopologyConf/taskName"

  22.         if (zookeeper.exists(node2, null) != null) {

  23.             List list = optzk.getNodeChild(zookeeper, node2);

  24.             for (String chNode : list) {

  25.                 optzk.deleteNode(zookeeper, node2 + StrTmp + chNode);

  26.             }

  27.         } else {

  28.             optzk.createNode(zookeeper, node2, "taskName");

  29.             System.out.println("TaskName");

  30.         }

  31.         // 获取MetaSpout并写入zk配置服务器中

  32.         String SpoutZkConnect = pros.getProperty("spout.zkConnect",

  33.                 "localhost:2181");

  34.         String SpoutZkRoot = pros.getProperty("spout.zkRoot", "/meta");

  35.         String SpoutZkTopic = pros.getProperty("spout.zkTopic", "storm_test");

  36.         String SpoutZkGroup = pros.getProperty("spout.zkGroup", "storm_test");

  37.         optzk.createNode(zookeeper, node2 + StrTmp + "spout.zkConnect",

  38.                 SpoutZkConnect);

  39.         optzk.createNode(zookeeper, node2 + StrTmp + "spout.zkRoot",

  40.                 SpoutZkRoot);

  41.         optzk.createNode(zookeeper, node2 + StrTmp + "spout.zkTopic",

  42.                 SpoutZkTopic);

  43.         optzk.createNode(zookeeper, node2 + StrTmp + "spout.zkGroup",

  44.                 SpoutZkGroup);

  45.         // 获取MetaBolt并写入zk配置服务器中

  46.         String BoltZkConnect = pros.getProperty("bolt.zkConnect",

  47.                 "localhost:2181");

  48.         String BoltZkRoot = pros.getProperty("bolt.zkRoot", "/meta");

  49.         String BoltZkTopic = pros.getProperty("bolt.zkTopic", "storm_test");

  50.         optzk.createNode(zookeeper, node2 + StrTmp + "bolt.zkConnect",

  51.                 BoltZkConnect);

  52.         optzk.createNode(zookeeper, node2 + StrTmp + "bolt.zkRoot", BoltZkRoot);

  53.         optzk.createNode(zookeeper, node2 + StrTmp + "bolt.zkTopic",

  54.                 BoltZkTopic);

  55.         // 获取MonitorBolt参数并写入zk配置服务器中

  56.         String MatchLogic = pros.getProperty("match.logic", "AND");

  57.         String MatchType = pros.getProperty("match.type",

  58.                 "regular::range::routine0");

  59.         String MatchField = pros.getProperty("match.field", "1::2::5");

  60.         String MatchFieldValue = pros.getProperty("match.fieldValue",

  61.                 ".*google.*::200,2001::ina");

  62.         optzk.createNode(zookeeper, node2 + StrTmp + "match.logic", MatchLogic);

  63.         optzk.createNode(zookeeper, node2 + StrTmp + "match.type", MatchType);

  64.         optzk.createNode(zookeeper, node2 + StrTmp + "match.field", MatchField);

  65.         optzk.createNode(zookeeper, node2 + StrTmp + "match.fieldValue",

  66.                 MatchFieldValue);

  67.         // 获取MysqlBolt参数并写入zk配置服务器中

  68.         String MysqlHost = pros.getProperty("mysql.host_port",

  69.                 "192.168.2.240:3306");

  70.         String MysqlDatabase = pros.getProperty("mysql.database", "storm");

  71.         String MysqlFrom = pros.getProperty("mysql.from", "monitor");

  72.         String MysqlUsername = pros.getProperty("mysql.username", "root");

  73.         String MysqlPassword = pros.getProperty("mysql.password", "123456");

  74.         optzk.createNode(zookeeper, node2 + StrTmp + "mysql.host_port",

  75.                 MysqlHost);

  76.         optzk.createNode(zookeeper, node2 + StrTmp + "mysql.database",

  77.                 MysqlDatabase);

  78.         optzk.createNode(zookeeper, node2 + StrTmp + "mysql.from", MysqlFrom);

  79.         optzk.createNode(zookeeper, node2 + StrTmp + "mysql.username",

  80.                 MysqlUsername);

  81.         optzk.createNode(zookeeper, node2 + StrTmp + "mysql.password",

  82.                 MysqlPassword);

  83.         optzk.close(zookeeper);

  84.     }
复制代码


2.4.3 storm.conf.PutConf.java

该类为主要的申请节点，往zookeeper中写入配置信息，该类供主类调用，在nimbus执行时，将配置信息写入zk中。

2.4.4 storm.conf.GetConf.java

该类为从zookeeper中获取配置信息的类，为各个bolt调用，从节点中获取配置信息。
  1.   public boolean getZkConf(String configFile, String taskName)

  2.             throws Exception {

  3.         Properties pros = new Properties();

  4.         File config = new File(configFile);

  5.         pros.load(new FileInputStream(config));

  6.         // 从nimbus域中获取配置文件zk服务器地址并连接zk

  7.         String nimbusZkConect = pros.getProperty("nimbus.zkConnect",

  8.                 "localhost:2181");

  9.         zookeeper = optzk.connect(nimbusZkConect);

  10.         // 创建配置根目录

  11.         String ConfRoot = "TopologyConf";

  12.         String StrTmp = "/";

  13.         String node2 = StrTmp + ConfRoot + StrTmp + taskName;

  14.         // 创建二级根目录"/TopologyConf/taskName"

  15.         if (zookeeper.exists(node2, null) != null) {

  16.             //// 获取MetaSpout的配置信息

  17.             SpoutZkConnect = optzk.getData(zookeeper, node2 + StrTmp + "spout.zkConnect");

  18.             SpoutZkRoot = optzk.getData(zookeeper, node2 + StrTmp + "spout.zkRoot");

  19.             SpoutZkTopic = optzk.getData(zookeeper, node2 + StrTmp + "spout.zkTopic");

  20.             SpoutZkGroup = optzk.getData(zookeeper, node2 + StrTmp + "spout.zkGroup");

  21.             // 获取MetaBolt的配置信息

  22.             BoltZkConnect = optzk.getData(zookeeper, node2 + StrTmp + "bolt.zkConnect");

  23.             BoltZkRoot = optzk.getData(zookeeper, node2 + StrTmp + "bolt.zkRoot");

  24.             BoltZkTopic = optzk.getData(zookeeper, node2 + StrTmp + "bolt.zkTopic");

  25.             // 获取MonitorBolt的配置信息

  26.             MatchLogic = optzk.getData(zookeeper, node2 + StrTmp + "match.logic");

  27.             MatchType = optzk.getData(zookeeper, node2 + StrTmp + "match.type");

  28.             MatchField = optzk.getData(zookeeper, node2 + StrTmp + "match.field");

  29.             MatchFieldValue = optzk.getData(zookeeper, node2 + StrTmp + "match.fieldValue");

  30.             // 获取MysqlBolt的配置信息

  31.             MysqlHost = optzk.getData(zookeeper, node2 + StrTmp + "mysql.host_port");

  32.             MysqlDatabase = optzk.getData(zookeeper, node2 + StrTmp + "mysql.database");

  33.             MysqlFrom = optzk.getData(zookeeper, node2 + StrTmp + "mysql.from");

  34.             MysqlUsername = optzk.getData(zookeeper, node2 + StrTmp + "mysql.username");

  35.             MysqlPassword = optzk.getData(zookeeper, node2 + StrTmp + "mysql.password");

  36.             optzk.close(zookeeper);

  37.             return true;

  38.         } else {

  39.             System.out.println("Conf is not exist!");

  40.             return false;

  41.         }

  42.     }
复制代码



3 Zookeeper相关API

方法名称描述
String create(final String path, byte data[], List acl, CreateMode createMode)创建一个znode节点，
参数： 路径、 znode内容，ACL(访问控制列表)、 znode创建类型
void delete(final String path, int version)删除一个znode节点，
参数： 路径、版本号；如果版本号与znode的版本号不一致，将无法删除，是一种乐观加锁机制；如果将版本号设置为-1，不会去检测版本，直接删除；
Stat exists(final String path, Watcher watcher)判断某个znode节点是否存在
参数： 路径、Watcher(监视器)；当这个znode节点被改变时，将会触发当前Watcher
Stat exists(String path, boolean watch)判断某个znode节点是否存在
参数： 路径、并设置是否监控这个目录节点，这里的 watcher 是在创建 ZooKeeper 实例时指定的 watcher
Stat setData(final String path, byte data[], int version)设置某个znode上的数据
参数： 路径、数据、版本号；如果为-1，跳过版本检查
byte[] getData(final String path, Watcher watcher, Stat stat)获取某个znode上的数据
参数： 路径、监视器、数据版本等信息
List getChildren(final String path, Watcher watcher)获取某个节点下的所有子节点
参数： 路径、监视器;该方法有多个重载


4 项目拓展

MonitorTopology目前已经完善了metaq与storm接口，实时过滤处理，数据存储方面实现了metaq重写接口，mysql存储接口，整个架构的统一配置系统。就总体实时处理架构来说，已经越来越完善。
项目扩展的目标：

(1) metaq数据获取，即尝试从现有前端动态日志中（模拟前端业务系统动态产生类似log4j日志）动态获取日志，并写入metaq中作为源数据；
(2) 丰富实时处理部分的bolt，增加其他功能bolt，如求Top N、条件统计、数据拆分等等；
(3) 丰富数据存储部分接口，特别是与hadoop接口；
(4) 使用zookeeper或者是mysql实现架构的元数据部分。

不积跬步无以至千里，我会一步一步把这个storm架构填充丰满。

PS：设计思路可以先参考参考，代码周末的时候调试调试，没问题放出来。

相关文档

Storm项目：流数据监控一流数据监控设计文档
http://www.aboutyun.com/thread-10042-1-1.html


Storm项目：流数据监控二：流数据监控代码详解
http://www.aboutyun.com/thread-10047-1-1.html

Storm项目：流数据监控三：流数据监控示例运行
http://www.aboutyun.com/thread-10046-1-1.html

Storm项目：流数据监控四---流数据监控MetaQ接口
http://www.aboutyun.com/thread-10045-1-1.html


实时处理方案架构 - Storm实时处理
http://www.aboutyun.com/thread-10043-1-1.html


本文地址：http://www.blogchong.com/post/storm_monitor_zkconfig_1576.html
回复

使用道具 举报

自定义广告语

已有(2)人评论

跳转到指定楼层
正序浏览
yidane 发表于 2014-11-19 14:24:55
写的非常清晰易懂,顶楼主
回复

使用道具 举报

笨笨_浪子 发表于 2014-11-21 20:01:07
真实感受了Zookeeper配置管理的应用。
回复

使用道具 举报

高级模式
B Color Image Link Quote Code Smilies
您需要登录后才可以回帖 登录 | 立即注册

本版积分规则

发表新帖
tntzbzc

高级会员

关注

  • 90

    主题

  • 666

    帖子

  • 19

    粉丝

TA的主题
关闭

推荐上一条 /2 下一条

关于我们· 联系我们· 加入我们· 小黑屋· 合作伙伴   

Copyright © 2001-2024 About云-梭伦科技 Powered by Discuz! X3.4 Licensed Discuz Team.

简书 / Smrz 京ICP备2020039040号 Wxb 简书网举报电话：021-34700000