案例演示
实时采集(监听目录)：Spool +File + HDFS
Spool 是Source来源于目录，有文件进入目录就摄取，File Channel将它暂存到磁盘，最终目的地是HDFS
即只要某个目录不断有文件，HDFS上也会同步到所有数据。
配置方案
[root@qianfeng01 flumeconf]# vi spool-hdfs.conf
a1.sources = r1
a1.channels = c1
a1.sinks = k1
a1.sources.r1.type = spooldir
a1.sources.r1.spoolDir = /root/flumedata/input/2020/01/
a1.channels.c1.type = file
a1.channels.c1.checkpointDir = /root/flumedata/checkpoint
a1.channels.c1.dataDirs = /root/flumedata/data
a1.sinks.k1.type = hdfs
a1.sinks.k1.hdfs.path = hdfs://qianfeng01:8020/flume/spooldir
a1.sinks.k1.hdfs.filePrefix =
a1.sinks.k1.hdfs.round = true
a1.sinks.k1.hdfs.roundValue = 10
a1.sinks.k1.hdfs.roundUnit = minute
a1.sinks.k1.hdfs.fileSuffix= .log
a1.sinks.k1.hdfs.rollInterval=60
a1.sinks.k1.hdfs.fileType=DataStream
a1.sinks.k1.hdfs.writeFormat=Text
a1.sources.r1.channels = c1
a1.sinks.k1.channel = c1
复制代码启动Agent
[root@qianfeng01 flumeconf]# flume-ng agent -c ../conf/ -f ./spool-hdfs.conf -n a1 -Dflume.root.logger=INFO,console
复制代码测试
--1. 向/home/flume/input/2020/01/目录里添加几个文件
[root@qianfeng01 ~]# cd /home/flume/input/2020/01/
[root@qianfeng01 01]# echo "hello world" >>a1.log
[root@qianfeng01 01]# echo "no zuo no die" >>a2.log
[root@qianfeng01 01]# echo "ao li gei" >>a3.sh
--2. 查看hdfs的目录内容
--3. 在查看一下home/flume/input/2020/01/目录里的内容
[root@qianfeng01 01]# ll
复制代码注意：采集文件夹内的数据文件不能有同名文件及不能修改已经采集的文件内容
实时采集(监听目录)：Spool+ mem+logger
Spool: Source来源于目录，有文件进入目录就摄取，File Channel将它暂存到磁盘，最终目的地是HDFS
即只要某个目录不断有文件，HDFS上也会同步到所有数据。mem:通过内存来传输数据
logger:是传送数据到日志
配置方案
[root@qianfeng01 flumeconf]# vi spool-logger.conf
a1.sources = r1
a1.channels = c1
a1.sinks = s1
a1.sources.r1.type=spooldir
a1.sources.r1.spoolDir = /root/flumedata/spool
#设置了前缀的代表已经采集完成
a1.sources.r1.fileSuffix = .COMPLETED
#never代表采集完不删除
a1.sources.r1.deletePolicy=never
#是否添加header中的文件绝对路径显示,false代表不显示
a1.sources.r1.fileHeader=false
#文件路径对应的fileHeaderKey
a1.sources.r1.fileHeaderKey=file
#是否添加header中的文件名字,false代表不显示
a1.sources.r1.basenameHeader=false
#文件名字对用的basenameHeaderKey
a1.sources.r1.basenameHeaderKey=basename
a1.sources.r1.batchSize=100
a1.sources.r1.inputCharset=UTF-8
a1.sources.r1.bufferMaxLines=1000
a1.channels.c1.type=memory
a1.channels.c1.capacity=1000
a1.channels.c1.transactionCapacity=100
a1.channels.c1.keep-alive=3
a1.channels.c1.byteCapacityBufferPercentage=20
a1.channels.c1.byteCapacity=800000
a1.sinks.s1.type=logger
a1.sinks.s1.maxBytesToLog=16
a1.sources.r1.channels=c1
a1.sinks.s1.channel=c1
复制代码启动Agent
[root@qianfeng01 flumeconf]# flume-ng agent -c ../conf -f ./spool-logger.conf -n a1 -Dflume.root.logger=INFO,console
复制代码测试
[root@qianfeng01 ~]# for i in `seq 1 10` ; do echo $i >> /root/flume/spool/$i;done
复制代码案例演示：HTTP+Memory+Logger
http: 表示数据来源是http网络协议,一般接收的请求为get或post请求. 所有的http请求会通过插件格式的Handle转化为一个Flume的Event数据.
mem:表示用内存传输通道
logger:表示输出格式为Logger格式
配置方案
[root@qianfeng01 flumeconf]# vi http-logger.conf
a1.sources = r1
a1.channels = c1
a1.sinks = s1
a1.sources.r1.type=http
a1.sources.r1.bind = qianfeng01
a1.sources.r1.port = 6666
a1.sources.r1.handler = org.apache.flume.source.http.JSONHandler
a1.sources.r1.handler.nickname = JSON props
a1.channels.c1.type=memory
a1.channels.c1.capacity=1000
a1.channels.c1.transactionCapacity=100
a1.channels.c1.keep-alive=3
a1.channels.c1.byteCapacityBufferPercentage=20
a1.channels.c1.byteCapacity=800000
a1.sinks.s1.type=logger
a1.sinks.s1.maxBytesToLog=16
a1.sources.r1.channels=c1
a1.sinks.s1.channel=c1
复制代码启动agent的服务
[root@qianfeng01 ~]# flume-ng agent -c ../conf -f ./http-logger.conf -n a1 -Dflume.root.logger=INFO,console
复制代码测试
curl -X 指定请求方式 -d 指定发送内容 服务器地址
[root@qianfeng01 ~]# curl -X POST -d '[{"headers":{"name":"zhangsan","pwd":"123456"},"body":"this is my content"}]' http://qianfeng01:6666
