时间戳拦截器
Timestamp.conf
#1.定义agent名， source、channel、sink的名称 a4.sources = r1 a4.channels = c1 a4.sinks = k1 #2.具体定义source a4.sources.r1.type = spooldir a4.sources.r1.spoolDir = /opt/module/flume-1.8.0/upload #定义拦截器，为文件最后添加时间戳 a4.sources.r1.interceptors = i1 a4.sources.r1.interceptors.i1.type = org.apache.flume.interceptor.TimestampInterceptor$Builder #具体定义channel a4.channels.c1.type = memory a4.channels.c1.capacity = 10000 a4.channels.c1.transactionCapacity = 100 #具体定义sink a4.sinks.k1.type = hdfs a4.sinks.k1.hdfs.path = hdfs://bigdata166:9000/flume-interceptors/%H a4.sinks.k1.hdfs.filePrefix = events- a4.sinks.k1.hdfs.fileType = DataStream #不按照条数生成文件 a4.sinks.k1.hdfs.rollCount = 0 #HDFS上的文件达到128M时生成一个文件 a4.sinks.k1.hdfs.rollSize = 134217728 #HDFS上的文件达到60秒生成一个文件 a4.sinks.k1.hdfs.rollInterval = 60 #组装source、channel、sink a4.sources.r1.channels = c1 a4.sinks.k1.channel = c1
/opt/module/flume-1.8.0/bin/flume-ng agent -n a4 \ -f /opt/module/flume-1.8.0/jobconf/Timestamp.conf \ -c /opt/module/flume-1.8.0/conf \ -Dflume.root.logger=INFO,console
主机名拦截器
Host.conf
#1.定义agent
a1.sources= r1
a1.sinks = k1
a1.channels = c1
#2.定义source
a1.sources.r1.type = exec
a1.sources.r1.channels = c1
a1.sources.r1.command = tail -F /opt/Shalltest
#拦截器
a1.sources.r1.interceptors = i1
a1.sources.r1.interceptors.i1.type = host
#参数为true时用IP192.168.1.111，参数为false时用主机名，默认为true
a1.sources.r1.interceptors.i1.useIP = false
a1.sources.r1.interceptors.i1.hostHeader = agentHost
#3.定义sinks
a1.sinks.k1.type=hdfs
a1.sinks.k1.channel = c1
a1.sinks.k1.hdfs.path = hdfs://bigdata166:9000/flumehost/%H
a1.sinks.k1.hdfs.filePrefix = Shall_%{agentHost}
#往生成的文件加后缀名.log
a1.sinks.k1.hdfs.fileSuffix = .log
a1.sinks.k1.hdfs.fileType = DataStream
a1.sinks.k1.hdfs.writeFormat = Text
a1.sinks.k1.hdfs.rollInterval = 10
a1.sinks.k1.hdfs.useLocalTimeStamp = true
a1.channels.c1.type = memory
a1.channels.c1.capacity = 1000
a1.channels.c1.transactionCapacity = 100
a1.sources.r1.channels = c1
a1.sinks.k1.channel = c1
bin/flume-ng agent -c conf/ -f jobconf/Host.conf -n a1 -Dflume.root.logger=INFO,console
UUID拦截器
uuid.conf
a1.sources = r1 a1.sinks = k1 a1.channels = c1 a1.sources.r1.type = exec a1.sources.r1.channels = c1 a1.sources.r1.command = tail -F /opt/Shalltest a1.sources.r1.interceptors = i1 #type的参数不能写成uuid，得写具体，否则找不到类 a1.sources.r1.interceptors.i1.type = org.apache.flume.sink.solr.morphline.UUIDInterceptor$Builder #如果UUID头已经存在,它应该保存 a1.sources.r1.interceptors.i1.preserveExisting = true a1.sources.r1.interceptors.i1.prefix = UUID_ #如果sink类型改为HDFS，那么在HDFS的文本中没有headers的信息数据 a1.sinks.k1.type = logger a1.channels.c1.type = memory a1.channels.c1.capacity = 1000 a1.channels.c1.transactionCapacity = 100 a1.sources.r1.channels = c1 a1.sinks.k1.channel = c1
bin/flume-ng agent -c conf/ -f jobconf/uuid.conf -n a1 -Dflume.root.logger==INFO,console
