如何提高水槽水槽性能

Question

我通过水槽接收器将事件从水槽发送到卡夫卡主题....为此，我正在使用文件通道，但在写入卡夫卡时性能非常低，水槽接收器以每秒 190 条消息的速率写入而水槽源正在以每秒 3000 msg 的速率将事件转储到通道....我想提高我的接收器性能......请帮助我如何实现......我尝试了各种为那个配置....这是我的配置文件-----

agent1.sources = AspectJ
agent1.channels = fileTailChannel
agent1.sinks = APMNullSink

#AspectJ source
agent1.sources.AspectJ.type=com.flume.test.DumbSource
agent1.sources.AspectJ.path=/media/sf_New_Log/calltracedata.txt
agent1.sources.AspectJ.eventtype=CallTrace
agent1.sources.AspectJ.channels=fileTailChannel
agent1.sources.AspectJ.batchSize=1000
agent1.sources.AspectJ.batchDurationMillis=1000
agent1.sources.AspectJ.application = AspectJ
agent1.sources.AspectJ.multi.line.mode=true
agent1.sources.AspectJ.new.event.marker={
agent1.sources.AspectJ.event.terminator=}
agent1.sources.AspectJ.tailer.start.reading.from.end = false


#File Tail Channel 
agent1.channels.fileTailChannel.type = file
agent1.channels.fileTailChannel.checkpointDir = /tmp/flume/filechannel/checkpoint
agent1.channels.fileTailChannel.dataDirs = /tmp/flume/filechannel/data
agent1.channels.fileTailChannel.transactionCapacity=100000

#APM Null Sink
agent1.sinks.APMNullSink.type = com.flume.test.TestJSON2KafkaSink
agent1.sinks.APMNullSink.channel = fileTailChannel
agent1.sinks.APMNullSink.testSize = 1000
agent1.sinks.APMNullSink.zk.connect=sandbox.hortonworks.com:2181
agent1.sinks.APMNullSink.metadata.broker.list=sandbox.hortonworks.com:6667
agent1.sinks.APMNullSink.topic=Test
agent1.sinks.APMNullSink.producer.type=async
agent1.sinks.APMNullSink.serializer.class=kafka.serializer.StringEncoder
agent1.sinks.APMNullSink.batch.num.messages=1000
agent1.sinks.APMNullSink.batchSize=50000
agent1.sinks.APMNullSink.batchDurationMillis=1000
agent1.sinks.APMNullSink.queue.buffering.max.ms=5000
agent1.sinks.APMNullSink.queue.buffering.max.messages=100000
agent1.sinks.APMNullSink.send.buffer.bytes=2097152
agent1.sinks.APMNullSink.compression.codec=snappy

Answer 1

我认为你的问题是一切都在一个盒子上运行，所以对文件通道的快速摄取正在影响 kafka，它需要做更多的工作来编写。

我建议两种选择：

1. 设置文件通道的容量以限制队列大小，因此它不会比 kafka 摄取的速度快。
2. 或使用 kafka 频道：https://flume.apache.org/FlumeUserGuide.html#kafka-channel。但是，在此选项中，主题中的消息被包装到 AvroFlumeEvent 中，因此订阅者需要使用该类进行反序列化。

我也不明白为什么你需要自己的 com.flume.test.TestJSON2KafkaSink 而不是 Flume 自带的。也许您的代码中存在一些性能问题。