flume-ng收集windows日志笔记

flume-ng收集windows下的日志并且使用SPILLABLEMEMORY 方式，日志实时收集使用spoolDirTailFile并且写入到rabbitmq

1.下载[b]spoolDirTailFile 用到的jar 地址https://github.com/ningg/flume-ng-extends-source[/b]

2.下载rabbitmq sink用到的 jar  地址https://github.com/aweber/rabbitmq-flume-plugin

3.配置 flume-ng-config

agent.channels = c1

agent.sources = spoolDirTailFile

agent.channels = c1

agent.channels.c1.type = SPILLABLEMEMORY

agent.channels.c1.memoryCapacity = 10000

agent.channels.c1.overflowCapacity = 1000000

agent.channels.c1.byteCapacity = 800000

agent.channels.c1.checkpointDir =C:\\log\\checkpoint

agent.channels.c1.dataDirs = C:\\log\\data

agent.sinks.k1.channel = c1

agent.sinks.k1.type =com.aweber.flume.sink.rabbitmq.RabbitMQSink

agent.sinks.k1.host = 主机地址

agent.sinks.k1.port = 5672

agent.sinks.k1.virtual-host =mq-host

agent.sinks.k1.username = 用户名

agent.sinks.k1.password = 密码

agent.sinks.k1.exchange = mq-队列名称

agent.sinks.k1.routing-key = mq-key

agent.sinks.k1.publisher-confirms = true

agent.sinks=k1

# Spooling dir and tail file Source 

agent.sources.spoolDirTailFile.type = com.github.ningg.flume.source.SpoolDirectoryTailFileSource

# on WIN plantform spoolDir should be format like: E:/program files/spoolDir

# Note: the value of spoolDir MUST NOT be surrounded by quotation marks.

agent.sources.spoolDirTailFile.spoolDir = C:\\log\\file

agent.sources.spoolDirTailFile.fileSuffix = .COMPLETED

agent.sources.spoolDirTailFile.deletePolicy = never 

agent.sources.spoolDirTailFile.ignorePattern = ^$

agent.sources.spoolDirTailFile.targetPattern = .*(\\d){10}.*

agent.sources.spoolDirTailFile.targetFilename = yyyyMMddhh

agent.sources.spoolDirTailFile.trackerDir = .flumespooltail

agent.sources.spoolDirTailFile.consumeOrder = oldest

agent.sources.spoolDirTailFile.batchSize = 100

agent.sources.spoolDirTailFile.inputCharset = UTF-8

agent.sources.spoolDirTailFile.decodeErrorPolicy = REPLACE

agent.sources.spoolDirTailFile.deserializer = LINE

agent.sources.spoolDirTailFile.channels= c1

agent.sources.spoolDirTailFile.fileHeader= true

以下是配置多个agent 供大家参考

agent2.channels = c2

agent2.sources = spoolDirTailFile2

agent2.channels.c2.type = memory

agent2.sinks.k2.channel = c2

agent2.sinks.k2.type = logger

agent2.sinks=k2

# Spooling dir and tail file Source 

agent2.sources.spoolDirTailFile2.type = com.github.ningg.flume.source.SpoolDirectoryTailFileSource

# on WIN plantform spoolDir should be format like: E:/program files/spoolDir

# Note: the value of spoolDir MUST NOT be surrounded by quotation marks.

agent2.sources.spoolDirTailFile2.spoolDir = C:\\log1

agent2.sources.spoolDirTailFile2.fileSuffix = .COMPLETED

agent2.sources.spoolDirTailFile2.deletePolicy = never 

agent2.sources.spoolDirTailFile2.ignorePattern = ^$

agent2.sources.spoolDirTailFile2.targetPattern = .*(\\d){4}-(\\d){2}-(\\d){2}.*

agent2.sources.spoolDirTailFile2.targetFilename = yyyy-MM-dd

agent2.sources.spoolDirTailFile2.trackerDir = .flumespooltail

agent2.sources.spoolDirTailFile2.consumeOrder = oldest

agent2.sources.spoolDirTailFile2.batchSize = 100

agent2.sources.spoolDirTailFile2.inputCharset = UTF-8

agent2.sources.spoolDirTailFile2.decodeErrorPolicy = REPLACE

agent2.sources.spoolDirTailFile2.deserializer = LINE

agent2.sources.spoolDirTailFile2.channels= c2

agent2.sources.spoolDirTailFile2.fileHeader= true



4.新建bat,windows下启动和linux下使用flume-ng 稍有不同

set FLUME_HOME=C:\apache-flume-1.6.0-bin

set JAVA="%JAVA_HOME%\bin\java.exe"

set JAVA_OPTS=-Xmx512m

set CONF=%FLUME_HOME%\conf\flume-conf.properties

set AGENT=agent

%JAVA%  %JAVA_OPTS%  -Dlog4j.configuration=file:\\\%FLUME_HOME%\conf\log4j.properties -cp "%FLUME_HOME%\lib\*" org.apache.flume.node.Application -f %FLUME_HOME%\conf\flume-conf.properties -n   agent

:: %JAVA%  %JAVA_OPTS%  -Dlog4j.configuration=file:\\\%FLUME_HOME%\conf\log4j.properties -cp "%FLUME_HOME%\lib\*" org.apache.flume.node.Application -f %FLUME_HOME%\conf\flume-conf.properties -n   agent2

最后附上一些搜集的使用心得：

一、关于Source：
　　1、spool-source：适合静态文件，即文件本身不是动态变化的；
　　2、avro source可以适当提高线程数量来提高此source性能；
　　3、ThriftSource在使用时有个问题需要注意，使用批量操作时出现异常并不会打印异常内容而是'Thrift source %s could not append events to the channel.'，这是因为源码中在出现异常时，它并未捕获异常而是获取组件名称，这是源码中的一个bug，也可以说明thrift很少有人用，否则这个问题也不会存在在很多版本中；
　　4、如果一个source对应多个channel，默认就是每个channel是同样的一份数据，会把这批数据复制N份发送到N个channel中，所以如果某个channel满了会影响整体的速度的哦；
　　5、ExecSource官方文档已经说明是异步的，可能会丢数据哦，尽量使用tail -F，注意是大写的；
二、关于Channel：
　　1、采集节点建议使用新的复合类型的SpillableMemoryChannel，汇总节点建议采用memory channel，具体还要看实际的数据量，一般每分钟数据量超过120MB大小的flume
agent都建议用memory channel(自己测的file channel处理速率大概是2M/s，不同机器、不同环境可能不同，这里只提供参考)，因为一旦此agent的channel出现溢出情况，将会导致大多数时间处于file channel(SpillableMemoryChannel本身是file channel的一个子类，而且复合channel会保证一定的event的顺序的使得读完内存中的数据后，再需要把溢出的拿走，可能这时内存已满又会溢出。。。)，性能大大降低，汇总一旦成为这样后果可想而知；
　　2、调整memory 占用物理内存空间，需要两个参数byteCapacityBufferPercentage(默认是20)和byteCapacity(默认是JVM最大可用内存的0.8)来控制，计算公式是：byteCapacity = (int)((context.getLong('byteCapacity',
defaultByteCapacity).longValue() * (1 - byteCapacityBufferPercentage * .01 )) /byteCapacitySlotSize)，很明显可以调节这两个参数来控制，至于byteCapacitySlotSize默认是100，将物理内存转换成槽(slot)数，这样易于管理，但是可能会浪费空间，至少我是这样想的。。。；
　　3、还有一个有用的参数'keep-alive'这个参数用来控制channel满时影响source的发送，channel空时影响sink的消费，就是等待时间，默认是3s，超过这个时间就甩异常，一般不需配置，但是有些情况很有用，比如你得场景是每分钟开头集中发一次数据，这时每分钟的开头量可能比较大，后面会越来越小，这时你可以调大这个参数，不至于出现channel满了得情况；
三、关于Sink：
　　1、avro sink的batch-size可以设置大一点，默认是100，增大会减少RPC次数，提高性能；
　　2、内置hdfs sink的解析时间戳来设置目录或者文件前缀非常损耗性能，因为是基于正则来匹配的，可以通过修改源码来替换解析时间功能来极大提升性能，稍后我会写一篇文章来专门说明这个问题；
　　3、RollingFileSink文件名不能自定义，而且不能定时滚动文件，只能按时间间隔滚动，可以自己定义sink，来做定时写文件；
　　4、hdfs sink的文件名中的时间戳部分不能省去，可增加前缀、后缀以及正在写的文件的前后缀等信息；'hdfs.idleTimeout'这个参数很有意义，指的是正在写的hdfs文件多长时间不更新就关闭文件，建议都配置上，比如你设置了解析时间戳存不同的目录、文件名，而且rollInterval=0、rollCount=0、rollSize=1000000，如果这个时间内的数据量达不到rollSiz
ad01
e的要求而且后续的写入新的文件中了，就是一直打开，类似情景不注意的话可能很多；'hdfs.callTimeout'这个参数指的是每个hdfs操作(读、写、打开、关闭等)规定的最长操作时间，每个操作都会放入'hdfs.threadsPoolSize'指定的线程池中得一个线程来操作；
　　5、关于HBase sink(非异步hbase sink：AsyncHBaseSink)，rowkey不能自定义，而且一个serializer只能写一列，一个serializer按正则匹配多个列，性能可能存在问题，建议自己根据需求写一个hbase sink；
　　6、avro sink可以配置failover和loadbalance，所用的组件和sinkgroup中的是一样的，而且也可以在此配置压缩选项，需要在avro source中配置解压缩；
四、关于SinkGroup：
　　1、不管是loadbalance或者是failover的多个sink需要共用一个channel；
　　2、loadbalance的多个sink如果都是直接输出到同一种设备，比如都是hdfs，性能并不会有明显增加，因为sinkgroup是单线程的它的process方法会轮流调用每个sink去channel中take数据，并确保处理正确，使得是顺序操作的，但是如果是发送到下一级的flume agent就不一样了，take操作是顺序的，但是下一级agent的写入操作是并行的，所以肯定是快的；
　　3、其实用loadbalance在一定意义上可以起到failover的作用，生产环境量大建议loadbalance；
五、关于监控monitor：
　　1、监控我这边做得还是比较少的，但是目前已知的有以下几种吧：cloudera manager（前提是你得安装CDH版本）、ganglia(这个天生就是支持的)、http(其实就是将统计信息jmx信息，封装成json串，使用jetty展示在浏览器中而已)、再一个就是自己实现收集监控信息，自己做(可以收集http的信息或者自己实现相应的接口实现自己的逻辑，具体可以参考我以前的博客)；
　　2、简单说一下cloudera manager这种监控，最近在使用，确实很强大，可以查看实时的channel进出数据速率、channel实时容量、sink的出速率、source的入速率等等，图形化的东西确实很丰富很直观，可以提供很多flume
agent整体运行情况的信息和潜在的一些信息；
六、关于flume启动：
　　1、flume组件启动顺序：channels——>sinks——>sources，关闭顺序：sources——>sinks——>channels；
　　2、自动加载配置文件功能，会先关闭所有组件，再重启所有组件；
　　3、关于AbstractConfigurationProvider中的Map<Class<? extends Channel>, Map<String, Channel>> channelCache这个对象，始终存储着agent中得所有channel对象，因为在动态加载时，channel中可能还有未消费完的数据，但是需要对channel重新配置，所以用以来缓存channel对象的所有数据及配置信息；
　　4、通过在启动命令中添加'no-reload-conf'参数为true来取消自动加载配置文件功能；
七、关于interceptor：
　　请看我的关于这个组件的博客，传送门；
八、关于自定义组件：sink、source、channel：
　　1、channel不建议自定义哦，这个要求比较高，其他俩都是框架式的开发，往指定的方法填充自己配置、启动、关闭、业务逻辑即可，以后有机会单独写一篇文章来介绍；
　　2、关于自定义组件请相信github，上面好多好多好多，可以直接用的自定义组件....；