Flume KafkaChannel 跨分区平衡消息

Question

我已经设置了一个 flume 代理，它使用 SpoolDir 命令从目录中读取 CSV 文件。

我正在使用通道类型 KafkaChannel，以便将这些消息推送到具有 10 个分区的 Kafka 主题，稍后可以由 Spark 应用程序处理。

我遇到的问题是每个文件都写入 Kafka 主题的单个分区。有些文件比其他文件大得多，这意味着消息在主题的分区中分布非常不均匀。这使得为我的 spark 执行器分配正确数量的资源变得异常困难，因为一些执行器最终完成了所有繁重的工作，而另一些则坐在那里等待将一些日志添加到他们的分区。

有什么方法可以配置 flume 中的 KafkaChannel 来平衡主题分区之间的消息，或者限制一次发送的消息数量，从而将负载分散到所有可用分区？

我在 flume 中试过以下配置选项但没有成功：

a1.channels.kafkaChannel.capacity = 100
a1.channels.kafkaChannel.transactionCapacity = 100
a1.channels.kafkaChannel.batch.size = 100

KafkaChannel 源代码已经过轻微修改以满足我的需要，但此处指定的默认配置选项仍然可用：http://flume.apache.org/FlumeUserGuide.html#kafka-channel

完整配置文件（我删除了主机名和其他关键信息）

a1.sources = src
a1.channels = kafkaChannel

a1.sources.src.type = spooldir
a1.sources.src.channels = kafkaChannel
a1.sources.src.spoolDir = /data/silk/flume/V5
a1.sources.src.fileHeader = true
a1.sources.src.trackerDir = .flumespool
a1.sources.src.consumeOrder = oldest
a1.sources.src.deletePolicy = immediate
a1.sources.src.decodeErrorPolicy = REPLACE
a1.sources.src.pollDelay = 12000

a1.channels.kafkaChannel.type = com.example.flume.channel.kafka.KafkaChannel
a1.channels.kafkaChannel.brokerList = host1:9092,host2:9092,host3:9092
a1.channels.kafkaChannel.topic = TEST-TOPIC
a1.channels.kafkaChannel.capacity = 100
a1.channels.c1.transactionCapacity = 100
a1.channels.kafkaChannel.zookeeperConnect = host1:2181,host2:2181,host3:2181
a1.channels.kafkaChannel.parseAsFlumeEvent = false

感谢任何帮助，提前致谢！

Answer 1

对于遇到此问题的其他人，我找到了解决方法：

通过实施 MemoryChannel 和 KafkaSink 而不是将日志直接推送到 KafkaChannel，我可以看到消息在我的 Kafka 主题分区之间更加均衡。

Flume KafkaChannel 跨分区平衡消息

Flume KafkaChannel balance messages across partitions

flume

apache-kafka