使用agent1监控文件变动agent1 将变动内容传递给agent2，agent2负责存储到HDFS，同时agent1将变内容传递给agent3
，agent3负责输出到LocalFileSystem
- 首先配置三个agent.conf 对SCK进行分配
agent1.conf
# Name the components on this agent a1.sources = r1 a1.sinks = k1 k2 a1.channels = c1 c2 # 将数据流复制给所有 channel a1.sources.r1.selector.type = replicating # source a1.sources.r1.type = exec a1.sources.r1.command = tail -F /root/test.log # sink # sink 端的 avro 是一个数据发送者 a1.sinks.k1.type = avro a1.sinks.k1.hostname = b02master a1.sinks.k1.port = 520 a1.sinks.k2.type = avro a1.sinks.k2.hostname = b02master a1.sinks.k2.port = 521 # channel a1.channels.c1.type = memory a1.channels.c1.capacity = 1000 a1.channels.c1.transactionCapacity = 100 a1.channels.c2.type = memory a1.channels.c2.capacity = 1000 a1.channels.c2.transactionCapacity = 100 # Bind the source and sink to the channel a1.sources.r1.channels = c1 c2 a1.sinks.k1.channel = c1 a1.sinks.k2.channel = c2
agent2.conf
# Name the components on this agent a2.sources = r1 a2.sinks = k1 a2.channels = c1 # source 端的 avro 是一个数据接收服务 a2.sources.r1.type = avro a2.sources.r1.bind = b02master a2.sources.r1.port = 520 # Describe the sink a2.sinks.k1.type = hdfs a2.sinks.k1.hdfs.path = hdfs://b02master:9000/flume/%Y%m%d/%H #上传文件的前缀 a2.sinks.k1.hdfs.filePrefix = agent2- #是否按照时间滚动文件夹 a2.sinks.k1.hdfs.round = true #多少时间单位创建一个新的文件夹 a2.sinks.k1.hdfs.roundValue = 1 #重新定义时间单位 a2.sinks.k1.hdfs.roundUnit = hour #是否使用本地时间戳 a2.sinks.k1.hdfs.useLocalTimeStamp = true #积攒多少个 Event 才 flush 到 HDFS 一次 a2.sinks.k1.hdfs.batchSize = 100 #设置文件类型，可支持压缩 a2.sinks.k1.hdfs.fileType = DataStream #多久生成一个新的文件 a2.sinks.k1.hdfs.rollInterval = 30 #设置每个文件的滚动大小大概是 128M a2.sinks.k1.hdfs.rollSize = 134217700 #文件的滚动与 Event 数量无关 a2.sinks.k1.hdfs.rollCount = 0 # Describe the channel a2.channels.c1.type = memory a2.channels.c1.capacity = 1000 a2.channels.c1.transactionCapacity = 100 # Bind the source and sink to the channel a2.sources.r1.channels = c1 a2.sinks.k1.channel = c1
agent3.conf
# Name the components on this agent a3.sources = r1 a3.sinks = k1 a3.channels = c2 # Describe/configure the source a3.sources.r1.type = avro a3.sources.r1.bind = b02master a3.sources.r1.port = 521 # Describe the sink a3.sinks.k1.type = file_roll a3.sinks.k1.sink.directory = /root/agent3 # Describe the channel a3.channels.c2.type = memory a3.channels.c2.capacity = 1000 a3.channels.c2.transactionCapacity = 100 # Bind the source and sink to the channel a3.sources.r1.channels = c2 a3.sinks.k1.channel = c2
这里的AvroSource被当作了服务端，所有应该先开启agent2和agent3
注意：输出的本地目录必须是已经存在的目录，如果该目录不存在，并不会创建新的目录
启动命令
bin/flume-ng agent -c conf -f conf/group1/agent1.conf -n a1 -Dflume.root.logger=INFO,console bin/flume-ng agent -c conf -f conf/group1/agent2.conf -n a2 -Dflume.root.logger=INFO,console bin/flume-ng agent -c conf -f conf/group1/agent3.conf -n a3 -Dflume.root.logger=INFO,console
在test.log 中追加数据 在group中和HDFS中分别去查看是否有对应的数据生成
