前提

https://blog.csdn.net/qq_41489540/article/details/109175329

编写Flume配置文件

f1.conf

#a1是agent的名称，a1中定义了一个叫r1的source，如果有多个，使用空格间隔 a1.sources = r1 a1.channels = c1 c2 #组名名.属性名=属性值 a1.sources.r1.type=TAILDIR a1.sources.r1.filegroups=f1 # 一批写多少个 a1.sources.r1.batchSize=1000 #读取/tmp/logs/app-yyyy-mm-dd.log ^代表以xxx开头$代表以什么结尾 .代表匹配任意字符 #+代表匹配任意位置 a1.sources.r1.filegroups.f1=/tmp/logs/^app.+.log$ #JSON文件的保存位置 a1.sources.r1.positionFile=/root/soft/apache-flume-1.7.0/custdata/log_position.json #定义拦截器 a1.sources.r1.interceptors = i1 a1.sources.r1.interceptors.i1.type = com.atguigu.dw.flume.MyInterceptor$Builder #定义ChannelSelector a1.sources.r1.selector.type = multiplexing a1.sources.r1.selector.header = topic # 要和你自己写的拦截器一样才行. a1.sources.r1.selector.mapping.topic_start = c1 a1.sources.r1.selector.mapping.topic_event = c2 #定义chanel为 KafkaChannel a1.channels.c1.type=org.apache.flume.channel.kafka.KafkaChannel # Kafka地址 a1.channels.c1.kafka.bootstrap.servers=zjj101:9092,zjj102:9092,zjj103:9092 #定义Kafka主题地址.如果不自己创建的话会自动创建,建议自己创建,自己创建可以指定分区和副本. a1.channels.c1.kafka.topic=topic_start # 不希望存header信息 a1.channels.c1.parseAsFlumeEvent=false a1.channels.c2.type=org.apache.flume.channel.kafka.KafkaChannel a1.channels.c2.kafka.bootstrap.servers=zjj101:9092,zjj102:9092,zjj103:9092 a1.channels.c2.kafka.topic=topic_event a1.channels.c2.parseAsFlumeEvent=false #连接组件 同一个source可以对接多个channel，一个sink只能从一个channel拿数据！ a1.sources.r1.channels=c1 c2

创建两个topic

创建 topic_start 和 topic_event ,都是三个分区两个副本数

启动Flume收集日志

[root@zjj101 conf]# flume-ng agent -c conf/ -n a1 -f /root/soft/apache-flume-1.7.0/conf/f1.conf -Dflume.root.logger=DEBUG,console

查看Flume记录日志的json文件

发现已经读了app-2020-10-15.log 文件.

[root@zjj101 custdata]# cat log_position.json [{"inode":1870589,"pos":696353,"file":"/tmp/logs/app-2020-10-15.log"}] [root@zjj101 custdata]#

用kafka 工具查看

发现已经有数据了