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

日志即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 = mysourceLogAgent.channels = mychannelLogAgent.sinks = mysinkLogAgent.sources.mysource.type = spooldirLogAgent.sources.mysource.channels = mychannelLogAgent.sources.mysource.spoolDir =/Users/wangsen/hadoop/apache-flume-1.7.0-bin/conf_copyLogAgent.sinks.mysink.channel = mychannelLogAgent.sinks.mysink.type = hdfsLogAgent.sinks.mysink.hdfs.path = hdfs://namenode:9000/data/logs2/LogAgent.sinks.mysink.hdfs.rollInterval = 30LogAgent.sinks.mysink.hdfs.batchSize = 10000LogAgent.sinks.mysink.hdfs.rollSize = 0LogAgent.sinks.mysink.hdfs.rollCount = 10000LogAgent.sinks.mysink.hdfs.fileType = DataStreamLogAgent.sinks.mysink.hdfs.useLocalTimeStamp = trueLogAgent.channels.mychannel.type = memoryLogAgent.channels.mychannel.capacity = 10000LogAgent.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

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

HDFS:

-rw-r--r-- 3 root supergroup 8567 2018-08-23 09:50 /data/logs2/FlumeData.1534989021404

生成一个文件，没有超过10000行就保存在一个文件。文件名称可以在.properties文件中配置。

总结

本文是Flume基本实验，TailDir是一种更强大的目录源Source，支持文件级的监听。通过设置Decoder可以文件作为事件（不以Line为Event），实现文件夹的同步。通过级联方式，实现多个主机之间高可靠文件/文件夹同步。