配置flume.conf

為我們的source channel sink起名

a1.sources = r1
a1.channels = c1
a1.sinks = k1

指定我們的source收集到的數據發送到哪個管道

a1.sources.r1.channels = c1

指定我們的source數據收集策略

a1.sources.r1.type = spooldir
a1.sources.r1.spoolDir = /export/servers/flumedata
a1.sources.r1.deletePolicy = never
a1.sources.r1.fileSuffix = .COMPLETED
a1.sources.r1.ignorePattern = ^(.)*\.tmp$
a1.sources.r1.inputCharset = GBK

指定我們的channel為memory,即表示所有的數據都裝進memory當中

a1.channels.c1.type = memory

指定我們的sink為kafka sink，并指定我們的sink從哪個channel當中讀取數據

a1.sinks.k1.channel = c1
a1.sinks.k1.type = org.apache.flume.sink.kafka.KafkaSink
a1.sinks.k1.kafka.topic = test
a1.sinks.k1.kafka.bootstrap.servers = node01:9092,node02:9092,node03:9092
a1.sinks.k1.kafka.flumeBatchSize = 20
a1.sinks.k1.kafka.producer.acks = 1

啟動flume

bin/flume-ng agent --conf conf --conf-file conf/flume.conf --name a1 -Dflume.root.logger=INFO,console

轉載于:https://www.cnblogs.com/alexzhang92/p/11100492.html

總結

以上是生活随笔為你收集整理的kafka和flume进行整合的日志采集的confi文件编写的全部內容，希望文章能夠幫你解決所遇到的問題。

如果覺得生活随笔網站內容還不錯，歡迎將生活随笔推薦給好友。