flume+kafka+hdfs:拆分消息

b1uwtaje  于 2021-06-02  发布在  Hadoop
关注(0)|答案(1)|浏览(476)

我使用以下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+数据,所以结果文件包含schema+数据+schema+数据,导致它是一个无效的.avro文件。
如何拆分avro事件以将不同的kafka消息拆分为每个消息写入不同的文件
谢谢您

bakd9h0s

bakd9h0s1#

一种方法:假设您将您的源kafka传入数据称为“sourcetopic”。您可以将自定义接收器注册到此“sourcetopic”。

<FlumeNodeRole>.sinks.<your-sink>.type =net.my.package.CustomSink

在customsink中,您可以编写一个方法来区分传入的消息,将其拆分,然后重新发送到不同的“destinationtopic”。这个“destinationtopic”现在可以作为文件序列化的新flume源。
有关管道衬砌Flume,请参阅以下链接:https://flume.apache.org/flumeuserguide.html

相关问题