我有以下 Flume 代理配置来从 kafka 源读取消息并将其写回 HDFS 接收器
tier1.sources = source1
tier 1.channels = channel1
tier1.sinks = sink1
tier1.sources.source1.type = org.apache.flume.source.kafka.KafkaSource
tier1.sources.source1.zookeeperConnect = 192.168.0.100:2181
tier1.sources.source1.topic = test
tier1.sources.source1.groupId = flume
tier1.sources.source1.channels = channel1
tier1.sources.source1.interceptors = i1
tier1.sources.source1.interceptors.i1.type = timestamp
tier1.sources.source1.kafka.consumer.timeout.ms = 100
tier1.channels.channel1.type = org.apache.flume.channel.kafka.KafkaChannel
tier1.channels.channel1.brokerList = 192.168.0.100:9092
tier1.channels.channel1.topic = test
tier1.channels.channel1.zookeeperConnect = 192.168.0.100:2181/kafka
tier1.channels.channel1.parseAsFlumeEvent = false
tier1.sinks.sink1.channel = channel1
tier1.sinks.sink1.type = hdfs
tier1.sinks.sink1.hdfs.writeFormat = Text
tier1.sinks.sink1.hdfs.fileType = DataStream
tier1.sinks.sink1.hdfs.filePrefix = test-kafka
tier1.sinks.sink1.hdfs.fileSufix = .avro
tier1.sinks.sink1.hdfs.useLocalTimeStamp = true
tier1.sinks.sink1.hdfs.path = /tmp/kafka/%y-%m-%d
tier1.sinks.sink1.hdfs.rollCount=0
tier1.sinks.sink1.hdfs.rollSize=0
kafka 消息内容是 avro 数据，如果每个轮询周期只有一条 kafka 消息到达，则该数据会正确序列化到文件中。
当两个kafka消息同一批到达时，它们被分组在同一个HDFS文件上，由于avro消息同时包含schema + data，因此结果文件包含schema + data + schema + data，导致它是无效的.avro文件。
如何拆分 avro 事件以将不同的 kafka 消息拆分为将每个消息写入不同的文件中
谢谢你
一种方法：假设您将源 kafka 传入数据称为“SourceTopic”。您可以向此“SourceTopic”注册自定义接收器。
<FlumeNodeRole>.sinks.<your-sink>.type =net.my.package.CustomSink
在您的 CustomSink 中，您可以编写一个方法来区分传入消息，将其拆分，然后重新发送到不同的“DestinationTopic”。 这个“DestinationTopic”现在可以充当文件序列化的新水槽源。
管道内衬水槽请参考以下链接： https://flume.apache.org/FlumeUserGuide.html