Flume 日志收集系统 Spooldir-Source HDFS-sink

birdskyws

发布于 2018-09-12 15:00:59

1.6K0

发布于 2018-09-12 15:00:59

日志即log，记录发生的事件。以Nginx为例，有error_log和access_log 2个日志。access_log是访问日志，每条访问记录会产生几百字节的数据，随着访问量增加，日志文件会越来越大，必须定期清理日志。现在数据越来越重要，因此不能简单丢弃，要保存这些数据做更多数据分析。可以将数据保存到HDFS系统上，Flume是一个数据搬运软件，它扩展了很多功能，支持很多数据源。不编写代码利用Flume就可以搭建一个将log保存到HDFS的可靠系统。

一、Flume 组件

Source 采集信息源
Channel 消息缓存队列
Sink 从缓存队列中拉取消息，并处理。

消息 Record，Source封装Event(事件)成为Record对象，并保存到Channel中，Sink拉取Record并保存到目标系统中。

Sink处理完成之后，会向Channel发送确认消息，提供消息处理的可靠性。

因为Flume是一个大数据组件，在刚接触的时候犯了思维惯性错误，以为Source、Channel、Sink是部署在不同主机上的。如图一个Agent包括了三个组件，运行在一台主机上，准确的说一个JVM进程。常见的Source是agent可监听的文件夹、文件，Sink是hdfs。

二、配置文件

LogAgent.sources = mysource
LogAgent.channels = mychannel
LogAgent.sinks = mysink
LogAgent.sources.mysource.type = spooldir
LogAgent.sources.mysource.channels = mychannel
LogAgent.sources.mysource.spoolDir =/Users/wangsen/hadoop/apache-flume-1.7.0-bin/conf_copy
LogAgent.sinks.mysink.channel = mychannel
LogAgent.sinks.mysink.type = hdfs
LogAgent.sinks.mysink.hdfs.path = hdfs://namenode:9000/data/logs2/
LogAgent.sinks.mysink.hdfs.rollInterval = 30
LogAgent.sinks.mysink.hdfs.batchSize = 10000
LogAgent.sinks.mysink.hdfs.rollSize = 0
LogAgent.sinks.mysink.hdfs.rollCount = 10000
LogAgent.sinks.mysink.hdfs.fileType = DataStream
LogAgent.sinks.mysink.hdfs.useLocalTimeStamp = true
LogAgent.channels.mychannel.type = memory
LogAgent.channels.mychannel.capacity = 10000
LogAgent.channels.mychannel.transactionCapacity = 10000

运行flume bin/flume-ng agent --conf conf --conf-file conf/logagent.properties --name LogAgent -Dflume.root.logger=DEBUG,console

三、注意事项

1. sinks.mysink.hdfs.batchSize 和channels.mychannel.transactionCapacity

process failed org.apache.flume.ChannelException: Take list for MemoryTransaction, capacity 100 full, consider committing more frequently, increasing capacity, or in creasing thread count

如果sink.batchSize 大于 transactionCapacity：channel的处理能力被占满，得不到sink的确认消息，因为没有达到sink批处理数。

2.spooldir 监听目录中的文件

spooldir监听文件目录，当出现新文件时，将新文件转化成事件。默认deseriallizer的值是LINE，文件的每行封装成一个Event。因此，在sink端也是按代表一行的Record进行处理。

3.hdfs sink 配置

hdfs.fileType = DataStream ##保存文件时不用压缩 hdfs.rollCount = 10000 ##每个文件记录10000条Record，超过10000条分割文件 hdfs.rollSize = 0 ## 不以文件的大小分割 hdfs.batchSize = 10000 ## 批处理数，没达到时保存在.tmp文件中 hdfs.rollInterval = 30 ##批处理超时时间，将tmp文件写入到正式文件，并提交确认。

四、实验结果

源文件夹:

drwxr-xr-x 2 wangsen staff 64 8 23 09:50 .flumespool -rw-r--r-- 1 wangsen staff 1661 9 26 2016 flume-conf.properties.template.COMPLETED -rw-r--r-- 1 wangsen staff 1455 9 26 2016 flume-env.ps1.template.COMPLETED -rw-r--r-- 1 wangsen staff 1565 9 26 2016 flume-env.sh.template.COMPLETED -rw-r--r-- 1 wangsen staff 3107 9 26 2016 log4j.properties.COMPLETED -rw-r--r--@ 1 wangsen staff 778 8 23 09:49 logagent.properties.COMPLETED

处理成功数据，添加后缀.COMPLETED，此后缀可以在.properties文件中设置。 HDFS: