Flume怎么采集發(fā)送消息到kafka

這篇文章主要講解了“Flume怎么采集發(fā)送消息到kafka”，文中的講解內(nèi)容簡(jiǎn)單清晰，易于學(xué)習(xí)與理解，下面請(qǐng)大家跟著小編的思路慢慢深入，一起來(lái)研究和學(xué)習(xí)“Flume怎么采集發(fā)送消息到kafka”吧！

Flume agent腳本及配置如下

Mac

安裝：

brew install flume

啟動(dòng)腳本：

nohup /usr/local/Cellar/flume/1.9.0/bin/flume-ng agent --conf /data/flume/run/flume/conf/conf.d/ --conf-file /data/flume/run/flume/conf/conf.d/mg-res_test.conf --name mg-res_test -Dflume.codeadmin.logger=INFO,console &

linux

安裝：

http://flume.apache.org/download.html

啟動(dòng)腳本：

nohup /data/flume/run/flume/bin/flume-ng agent --conf /data/flume/run/flume/conf/conf.d/ --conf-file /data/flume/run/flume/conf/conf.d/mg-res_test.conf --name mg-res_test -Dflume.codeadmin.logger=INFO,console &

配置說(shuō)明

/data/flume/run/flume/conf/conf.d/mg-res_test.conf

flume主要配置信息，關(guān)注中文描述的部分即可

# Name the components on this agent
mg-res_test.sources = mg-res_test
mg-res_test.sinks = mg-res_test
mg-res_test.channels = mg-res_test
# Describe/configure the source
mg-res_test.sources.mg-res_test.type = TAILDIR
# flume 記錄偏移量文件
mg-res_test.sources.mg-res_test.positionFile=/data/logs/test/res-test/mg-res_test.json
mg-res_test.sources.mg-res_test.filegroups=f1
# 待采集的日志文件
mg-res_test.sources.mg-res_test.filegroups.f1=/data/logs/test/res-test/app.log
mg-res_test.sources.mg-res_test.fileHeader=true

# Describe the sink
mg-res_test.sinks.mg-res_test.channel = mg-res_test
mg-res_test.sinks.mg-res_test.type = org.apache.flume.sink.kafka.KafkaSink
# kafka topic配置 以實(shí)際為準(zhǔn)
mg-res_test.sinks.mg-res_test.kafka.topic = res-test
# kafka 配置信息 以實(shí)際為準(zhǔn)
mg-res_test.sinks.mg-res_test.kafka.bootstrap.servers = localhost:9092
mg-res_test.sinks.mg-res_test.kafka.flumeBatchSize = 2
mg-res_test.sinks.mg-res_test.kafka.producer.acks = 1
mg-res_test.sinks.mg-res_test.kafka.producer.linger.ms = 1
mg-res_test.sinks.mg-res_test.kafka.producer.compression.type = snappy
# Use a channel which buffers events in memory
mg-res_test.channels.mg-res_test.type = memory
mg-res_test.channels.mg-res_test.capacity = 100000
mg-res_test.channels.mg-res_test.transactionCapacity = 10000
# Bind the source and sink to the channel
mg-res_test.sources.mg-res_test.channels = mg-res_test
mg-res_test.sinks.mg-res_test.channel = mg-res_test

/data/flume/run/flume/conf/conf.d/flume-env.sh

配置java home即可

export JAVA_OPTS="-Xms128m -Xmx128m -Dcom.sun.management.jmxremote"
JAVA_HOME=/Library/Java/JavaVirtualMachines/jdk1.8.0_192.jdk/Contents/Home

驗(yàn)證

本地環(huán)境可以直接啟動(dòng)kafka消費(fèi)看有沒(méi)有消息進(jìn)入，測(cè)試環(huán)境請(qǐng)查詢kafka消息增量或業(yè)務(wù)系統(tǒng)實(shí)際消費(fèi)情況

kafka-console-consumer --bootstrap-server localhost:9092 --topic res-test --from-beginning

感謝各位的閱讀，以上就是“Flume怎么采集發(fā)送消息到kafka”的內(nèi)容了，經(jīng)過(guò)本文的學(xué)習(xí)后，相信大家對(duì)Flume怎么采集發(fā)送消息到kafka這一問(wèn)題有了更深刻的體會(huì)，具體使用情況還需要大家實(shí)踐驗(yàn)證。這里是億速云，小編將為大家推送更多相關(guān)知識(shí)點(diǎn)的文章，歡迎關(guān)注！