Kafka与avro记录

Question

我有以下:Source-Kafka topic(trans)Channel-memory Sink-Hdfs(avro _ event)

kafka主题trans中的数据是使用c#生产者编写的，并且有数千条avro记录。当我运行我的水槽消费者时，它开始将数据下沉到hdfs。问题是数据的格式是：模式数据模式数据

而不是:

模式数据数据

我猜这是因为flume需要一个带有{header} {body}的记录类型，而来自kafka的数据将只是{body}我知道有一种方法可以将写入主题的avro数据包装在avroFlumeEvent中，但这似乎不再是一个真正的avro记录，也许spark消费者或storm会更喜欢真正的avro中的数据。有没有一种方法可以处理这个主题，以便每次flume将数据滚动到hdfs时，数据都不用多个模式就可以写入？

Answer 1

一旦你将数据登陆Kafka，你是否考虑过使用LinkedIn的加缪。它将运行mapreduce作业，但您应该获得所需的模式数据数据布局。您还应该查看Confluent的kafka堆栈，特别是它提供的模式注册表和它提供的rest api。

Answer 2

我们实际上最终得到了这个工作。我们在C#生产者中使用microsoft .NET avro库而不是apache avro库。这意味着 avro 记录已正确序列化。我还需要将水槽接收器更改为使用“org.apache.flume.sink.hdfs.AvroEventSerializer$Builder”作为接收器序列化程序，而不是“avro_event”。我还需要包含一个连接到 kafka 源的 flume 拦截器，它将变量 “flume.avro.schema.url” 推送到水槽标头中，以便稍后由 hdfs sink 序列化器使用。

我看了一下加缪，但对于我们试图实现的东西来说，它似乎有点矫枉过正，一个连接到 kafka 主题的基本水槽通道，它将 avro 数据下沉到 hdfs。

我刚刚从构建水槽配置的java应用程序中撕下了拦截器位，希望它可以帮助遇到此问题的其他人：

                _flumeFileConfigProperties.put(_agentId+".sources." + _sourceId +".interceptors",_interceptorId);           
                _flumeFileConfigProperties.put(_agentId+".sources." + _sourceId + ".interceptors." + _interceptorId + ".type","static");
                _flumeFileConfigProperties.put(_agentId+".sources." + _sourceId + ".interceptors." + _interceptorId + ".key","flume.avro.schema.url");
                _flumeFileConfigProperties.put(_agentId+".sources." + _sourceId + ".interceptors." + _interceptorId + ".value",_avroProdSchemaLocation +_databaseName + "/" + _topic + "/record/" + _schemaVersion + "/" + _topicName + ".avsc");