日志即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:
-rw-r--r-- 3 root supergroup 8567 2018-08-23 09:50 /data/logs2/FlumeData.1534989021404
生成一个文件，没有超过10000行就保存在一个文件。文件名称可以在.properties文件中配置。
总结
本文是Flume基本实验，TailDir是一种更强大的目录源Source，支持文件级的监听。通过设置Decoder可以文件作为事件（不以Line为Event），实现文件夹的同步。通过级联方式，实现多个主机之间高可靠文件/文件夹同步。
网友评论