1． 案例场景
A、B 两台日志服务机器实时生产日志主要类型为 access.log、nginx.log、web.log
现在要求：
把 A、B 机器中的 access.log、nginx.log、web.log 采集汇总到 C 机器上然后统一收集到 hdfs 中。
但是在 hdfs 中要求的目录为：
/source/logs/access/20190101/**
/source/logs/nginx/20190101/**
/source/logs/web/20190101/**
2．场景分析
3．数据流程处理分析
4．功能实现
在服务器 node01 和服务器 node02 上
创建配置文件 exec_source_avro_sink.conf
# Name the components on this agenta1.sources = r1 r2 r3a1.sinks = k1a1.channels = c1
# Describe/configure the sourcea1.sources.r1.type = execa1.sources.r1.command = tail -F /root/data/access.loga1.sources.r1.interceptors = i1a1.sources.r1.interceptors.i1.type = static## static拦截器的功能就是往采集到的数据的header中插入自## 己定义的key-value对a1.sources.r1.interceptors.i1.key = typea1.sources.r1.interceptors.i1.value = access
a1.sources.r2.type = execa1.sources.r2.command = tail -F /root/data/nginx.loga1.sources.r2.interceptors = i2a1.sources.r2.interceptors.i2.type = statica1.sources.r2.interceptors.i2.key = typea1.sources.r2.interceptors.i2.value = nginx
a1.sources.r3.type = execa1.sources.r3.command = tail -F /root/data/web.loga1.sources.r3.interceptors = i3a1.sources.r3.interceptors.i3.type = statica1.sources.r3.interceptors.i3.key = typea1.sources.r3.interceptors.i3.value = web
# Describe the sinka1.sinks.k1.type = avroa1.sinks.k1.hostname = node03a1.sinks.k1.port = 41414
# Use a channel which buffers events in memorya1.channels.c1.type = memorya1.channels.c1.capacity = 20000a1.channels.c1.transactionCapacity = 10000
# Bind the source and sink to the channela1.sources.r1.channels = c1a1.sources.r2.channels = c1a1.sources.r3.channels = c1a1.sinks.k1.channel = c1
复制代码
在服务器 node03 上创建配置文件 avro_source_hdfs_sink.conf 文件内容为
#定义agent名， source、channel、sink的名称a1.sources = r1a1.sinks = k1a1.channels = c1
#定义sourcea1.sources.r1.type = avroa1.sources.r1.bind = node03a1.sources.r1.port =41414
#定义channelsa1.channels.c1.type = memorya1.channels.c1.capacity = 20000a1.channels.c1.transactionCapacity = 10000
#定义sinka1.sinks.k1.type = hdfsa1.sinks.k1.hdfs.path=hdfs://node01:8020/source/logs/%{type}/%Y%m%da1.sinks.k1.hdfs.filePrefix =eventsa1.sinks.k1.hdfs.fileType = DataStreama1.sinks.k1.hdfs.writeFormat = Text#时间类型a1.sinks.k1.hdfs.useLocalTimeStamp = true#生成的文件不按条数生成a1.sinks.k1.hdfs.rollCount = 0#生成的文件按时间生成a1.sinks.k1.hdfs.rollInterval = 30#生成的文件按大小生成a1.sinks.k1.hdfs.rollSize = 10485760#批量写入hdfs的个数a1.sinks.k1.hdfs.batchSize = 10000flume操作hdfs的线程数（包括新建，写入等）a1.sinks.k1.hdfs.threadsPoolSize=10#操作hdfs超时时间a1.sinks.k1.hdfs.callTimeout=30000
#组装source、channel、sinka1.sources.r1.channels = c1a1.sinks.k1.channel = c1
复制代码
配置完成之后，在服务器 node01 和 node02 上的/root/data 有数据文件 access.log、nginx.log、web.log。先启动服务器 node03 上的 flume，启动命令
在 flume 安装目录下执行 ：
bin/flume-ng agent -c conf -fconf/avro_source_hdfs_sink.conf -name a1 -Dflume.root.logger=DEBUG,console
然后在启动服务器上的 node01 和 node02，启动命令
在 flume 安装目录下执行 ：
bin/flume-ng agent -c conf -fconf/exec_source_avro_sink.conf -name a1 -Dflume.root.logger=DEBUG,console
评论