在《基于Flume的美团日志收集系统(一)架构和设计》中,我们详述了基于Flume的美团日志收集系统的架构设计,以及为什么做这样的设计。在本节中,我们将会讲述在实际部署和使用过程中遇到的问题,对Flume的功能改进和对系统做的优化。
1 Flume的问题总结
在Flume的使用过程中,遇到的主要问题如下:
a. Channel“水土不服”:使用固定大小的MemoryChannel在日志高峰时常报队列大小不够的异常;使用FileChannel又导致IO繁忙的问题;
b. HdfsSink的性能问题:使用HdfsSink向Hdfs写日志,在高峰时间速度较慢;
c. 系统的管理问题:配置升级,模块重启等;
2 Flume的功能改进和优化点
从上面的问题中可以看到,有一些需求是原生Flume无法满足的,因此,基于开源的Flume我们增加了许多功能,修改了一些Bug,并且进行一些调优。下面将对一些主要的方面做一些说明。
2.1 增加Zabbix monitor服务
一方面,Flume本身提供了http, ganglia的监控服务,而我们目前主要使用zabbix做监控。因此,我们为Flume添加了zabbix监控模块,和sa的监控服务无缝融合。
另一方面,净化Flume的metrics。只将我们需要的metrics发送给zabbix,避免 zabbix server造成压力。目前我们最为关心的是Flume能否及时把应用端发送过来的日志写到Hdfs上, 对应关注的metrics为:
- Source : 接收的event数和处理的event数
- Channel : Channel中拥堵的event数
- Sink : 已经处理的event数
2.2 为HdfsSink增加自动创建index功能
首先,我们的HdfsSink写到hadoop的文件采用lzo压缩存储。 HdfsSink可以读取hadoop配置文件中提供的编码类列表,然后通过配置的方式获取使用何种压缩编码,我们目前使用lzo压缩数据。采用lzo压缩而非bz2压缩,是基于以下测试数据:
event大小(Byte) | sink.batch-size | hdfs.batchSize | 压缩格式 | 总数据大小(G) | 耗时(s) | 平均events/s | 压缩后大小(G) |
---|---|---|---|---|---|---|---|
544 | 300 | 10000 | bz2 | 9.1 | 2448 | 6833 | 1.36 |
544 | 300 | 10000 | lzo | 9.1 | 612 | 27333 | 3.49 |
其次,我们的HdfsSink增加了创建lzo文件后自动创建index功能。Hadoop提供了对lzo创建索引,使得压缩文件是可切分的,这样Hadoop Job可以并行处理数据文件。HdfsSink本身lzo压缩,但写完lzo文件并不会建索引,我们在close文件之后添加了建索引功能。
1 /** 2 * Rename bucketPath file from .tmp to permanent location. 3 */ 4 private void renameBucket() throws IOException, InterruptedException { 5 if(bucketPath.equals(targetPath)) { 6 return; 7 } 8 9 final Path srcPath = new Path(bucketPath); 10 final Path dstPath = new Path(targetPath); 11 12 callWithTimeout(new CallRunner<Object>() { 13 @Override 14 public Object call() throws Exception { 15 if(fileSystem.exists(srcPath)) { // could block 16 LOG.info("Renaming " + srcPath + " to " + dstPath); 17 fileSystem.rename(srcPath, dstPath); // could block 18 19 //index the dstPath lzo file 20 if (codeC != null && ".lzo".equals(codeC.getDefaultExtension()) ) { 21 LzoIndexer lzoIndexer = new LzoIndexer(new Configuration()); 22 lzoIndexer.index(dstPath); 23 } 24 } 25 return null; 26 } 27 }); 28 }
2.3 增加HdfsSink的开关
我们在HdfsSink和DualChannel中增加开关,当开关打开的情况下,HdfsSink不再往Hdfs上写数据,并且数据只写向DualChannel中的FileChannel。以此策略来防止Hdfs的正常停机维护。
2.4 增加DualChannel
Flume本身提供了MemoryChannel和FileChannel。MemoryChannel处理速度快,但缓存大小有限,且没有持久化;FileChannel则刚好相反。我们希望利用两者的优势,在Sink处理速度够快,Channel没有缓存过多日志的时候,就使用MemoryChannel,当Sink处理速度跟不上,又需要Channel能够缓存下应用端发送过来的日志时,就使用FileChannel,由此我们开发了DualChannel,能够智能的在两个Channel之间切换。
其具体的逻辑如下:
1 /*** 2 * putToMemChannel indicate put event to memChannel or fileChannel 3 * takeFromMemChannel indicate take event from memChannel or fileChannel 4 * */ 5 private AtomicBoolean putToMemChannel = new AtomicBoolean(true); 6 private AtomicBoolean takeFromMemChannel = new AtomicBoolean(true); 7 8 void doPut(Event event) { 9 if (switchon && putToMemChannel.get()) { 10 //往memChannel中写数据 11 memTransaction.put(event); 12 13 if ( memChannel.isFull() || fileChannel.getQueueSize() > 100) { 14 putToMemChannel.set(false); 15 } 16 } else { 17 //往fileChannel中写数据 18 fileTransaction.put(event); 19 } 20 } 21 22 Event doTake() { 23 Event event = null; 24 if ( takeFromMemChannel.get() ) { 2