带有 avro 记录的 Kafka答案

【问题标题】：Kafka with avro records带有 avro 记录的 Kafka
【发布时间】：2015-05-01 12:12:46
【问题描述】：

我有以下几点：来源 - kafka 主题（反式）通道 - 内存接收器 - Hdfs (avro_event)

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

代替：

架构数据数据

我猜这是因为 Flume 期望使用 {header} {body} 的记录类型，而来自 kafka 的数据将只是 {body} 我知道有一种方法可以将写入主题的 avro 数据包装在 avroFlumeEvent 中，但是它似乎不再是真正的 avro 记录，也许火花消费者或风暴会更喜欢真正的 avro 中的数据.有没有办法处理这个主题，所以每次flume将数据滚动到hdfs时，数据都是在没有多个模式的情况下写入的？

【问题讨论】：

标签： hadoop hadoop-streaming apache-kafka flume avro

【解决方案1】：

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

我看过 camus，但对于我们试图实现的东西来说，它似乎有点矫枉过正，这是一个连接到 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");

【讨论】：

嗨@Colman 我被困在同一个问题即使在使用拦截器之后你还有代码的解决方案文件

【解决方案2】：

您是否考虑过使用来自 LinkedIn 的 Camus，一旦您将数据放到 kafka 上。它将运行 mapreduce 作业，但您应该获得所需的模式数据数据布局。您还应该查看 Confluent 的 kafka 堆栈，尤其是它提供的模式注册表和它提供的其余 api。

【讨论】：