Flume 採集傳送訊息到kafka
阿新 • • 發佈:2019-08-10
一些場景需要採集文字內容傳送到kafka,flume正好提供了這種解決方案
Flume agent指令碼及配置如下
Mac
安裝:
brew install flume
啟動指令碼:
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
啟動指令碼:
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 &
配置說明
/data/flume/run/flume/conf/conf.d/mg-res_test.conf
flume主要配置資訊,關注中文描述的部分即可
# 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配置 以實際為準 mg-res_test.sinks.mg-res_test.kafka.topic = res-test # kafka 配置資訊 以實際為準 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
驗證
本地環境可以直接啟動kafka消費看有沒有訊息進入,測試環境請查詢kafka訊息增量或業務系統實際消費情況
kafka-console-consumer --bootstrap-server localhost:9092 --topic res-test --fro