Flume采用zookeeper管理配置

rrt44

　　Flume支持通过zookeeper来管理Agent的配置，但是这是一个实验性的功能。配置文件必须先上传到zookeeper中。以下Agent在Zookeeper节点树的结构：
　　

- /flume
|- /a1 [Agent配置文件]
|- /a2 [Agent配置文件]
处理配置文件的类：　　
　　
　　

org.apache.flume.node.PollingZooKeeperConfigurationProvider ： 如果zookeeper指定的路径有变更，就从Zookeeper重新获取配置文件。

org.apache.flume.node.StaticZooKeeperConfigurationProvider ： 启动Flume后，不会重新加载配置文件，即使Zookeeper的配置文件有变更。

org.apache.flume.agent.embedded.MemoryConfigurationProvider ： 从存储中读取配置文件。传入数据格式是Map。

org.apache.flume.node.PollingPropertiesFileConfigurationProvider ： 定时冲硬盘读取配置文件。　　
　　


　　org.apache.flume.node.AbstractZooKeeperConfigurationProvider创建Zookeeper客户端：
　　

  protected CuratorFramework createClient() {
return CuratorFrameworkFactory.newClient(zkConnString,
new ExponentialBackoffRetry(1000, 1));
}
　　Flume采用Curator作为zookeeper的客户端，Curator是Netflix公司开源的一个Zookeeper客户端，与Zookeeper提供的原生客户端相比，Curator的抽象层次更高，简化了Zookeeper客户端的开发量。
　　


　　
　　Curator的maven配置：

<dependency>
<groupId>org.apache.curator</groupId>
<artifactId>apache-curator</artifactId>
<version>2.9.0</version>
</dependency>
　　


　　Zookeeper还有一个原生态的客户端，maven配置：

<dependency>
<groupId>org.apache.zookeeper</groupId>
<artifactId>zookeeper</artifactId>
<version>3.4.6</version>
</dependency>
　　
　　


　　使用原生态的客户端，上传配置文件flume-zookeeper.properties到zookeeper集群：
　　

@Test
public void uploadFileToZK() throws KeeperException, InterruptedException {
String propFilePath = &quot;D:\\flume-zookeeper.properties&quot;;
ZooKeeper zk = null;
try {
zk = new ZooKeeper(&quot;10.0.1.85:2181,10.0.1.86:2181,10.0.1.87:2181&quot;, 300000, new Watcher() {
// 监控所有被触发的事件
public void process(WatchedEvent event) {
System.out.println(&quot;已经触发了&quot; + event.getType() + &quot;事件！&quot;);
}
});
} catch (IOException e) {
// TODO Auto-generated catch block
e.printStackTrace();
}
if (zk.exists(&quot;/flume&quot;, true) == null) {
zk.create(&quot;/flume&quot;, null, Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
}
InputStream is = null;
ByteArrayOutputStream bytestream = null;
byte[] data = null;
try {
is = new FileInputStream(propFilePath);
bytestream = new ByteArrayOutputStream();
int ch;
while ((ch = is.read()) != -1) {
bytestream.write(ch);
}
data = bytestream.toByteArray();
System.out.println(new String(data));
} catch (FileNotFoundException e) {
e.printStackTrace();
} catch (IOException e) {
e.printStackTrace();
} finally {
try {
bytestream.close();
is.close();
} catch (IOException e) {
e.printStackTrace();
}
}
// 创建一个目录节点
Stat stat = zk.exists(&quot;/flume/a1&quot;, true);
if (stat == null) {
zk.create(&quot;/flume/a1&quot;, data, Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
} else {
zk.delete(&quot;/flume/a1&quot;, stat.getVersion());
zk.create(&quot;/flume/a1&quot;, data, Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
}
}
@Test
public void get() throws KeeperException, InterruptedException {
ZooKeeper zk = null;
try {
zk = new ZooKeeper(&quot;10.0.1.85:2181,10.0.1.86:2181,10.0.1.87:2181&quot;, 300000, new Watcher() {
// 监控所有被触发的事件
public void process(WatchedEvent event) {
System.out.println(&quot;已经触发了&quot; + event.getType() + &quot;事件！&quot;);
}
});
} catch (IOException e) {
e.printStackTrace();
}
System.out.println(new String(zk.getData(&quot;/flume/a1&quot;, true, null)));
}

flume-zookeeper.properties配置文件内容：　　
　　

a1.channels.c1.type = memory
a1.channels.c1.capacity = 1000
a1.channels.c1.transactionCapacity = 100
a1.sources.r1.channels = c1
a1.sources.r1.type = AVRO
a1.sources.r1.bind = 0.0.0.0
a1.sources.r1.port = 41414
a1.sinks.k1.channel = c1
a1.sinks.k1.type = file_roll
a1.sinks.k1.sink.directory = /logs/
a1.sinks.k1.sink.rollInterval = 0
a1.channels = c1
a1.sources = r1
a1.sinks = k1
　　
　　

avro获取数据，通过Memory Channel写入文件中。


配置文件上传到Zookeeper后，通过如下命令启动Flume：
$ bin/flume-ng agent –conf conf -z 10.0.1.85:2181,10.0.1.86:2181,10.0.1.87:2181 -p /flume –name a1 -Dflume.root.logger=INFO,console

参数名称
默认&#20540;
描述

z
–
Zookeeper的连接字符串hostname:port列表通过逗号分隔。
p
/flume
Agent配置文件的根路径。
zookeeper中的配置如果有更新，Flume会通过PollingZooKeeperConfigurationProvider类的refreshConfiguration方法重新加载配置：private void refreshConfiguration() throws IOException {
LOGGER.info(&quot;Refreshing configuration from ZooKeeper&quot;);
byte[] data = null;
ChildData childData = agentNodeCache.getCurrentData();
if (childData != null) {
data = childData.getData();
}
flumeConfiguration = configFromBytes(data);
eventBus.post(getConfiguration());
}