avro.io.AvroTypeException：数据 <avro 数据> 不是模式 {...} 的示例答案

【问题标题】：avro.io.AvroTypeException: The datum <avro data> is not an example of the schema {...}avro.io.AvroTypeException：数据 <avro 数据> 不是模式 {...} 的示例
【发布时间】：2016-06-13 06:17:53
【问题描述】：

我们正在努力将 Apache Storm 与 Confluent 框架与 Kafka 集成。我们正在使用一个名为“Pyleus”的风暴python包装器

我们设置了一个 Confluent-Kafka JDBC 连接器监控数据库表，每当数据库发生变化时，新记录将作为 Avro 格式的 Kafka 消息发送。

在 Pyleus bolt 中，我们能够获取 Kafka 消息，但是我们无法将其反序列化为 JSON。

我们正在使用两个名为“avro_json_serializer”和“avro”的 python-Avro 模块。当我尝试反序列化我放在一起的简单 Avro 文件时，它们会起作用。

Kafka 消息中 Avro 数据的 Avro 模式是使用 HTTP GET 从 Confluent 的模式注册表中获取的。我将 Kafka 消息中的架构和 Avro 数据放入两个文件中，这是我的测试程序：

import avro
import avro_json_serializer as ajs

import json

# Avro schema from Confluent's schema registry using HTTP GET
schema_string = open("realAvroSchemaFromKK.avsc").read()

schema_dict = json.loads(schema_string)
avro_schema = avro.schema.make_avsc_object(schema_dict, avro.schema.Names())

serializer = ajs.AvroJsonSerializer(avro_schema)

# Avro data with in Kafka message - I wrote it into this file
avrofile = open("realAvroFromKK.avro", "r")
avro = avrofile.read()

jsonData = serializer.to_json(avro) # where the code error out #

print jsonData

我解释错误消息的方式是我的 avro 架构不适合我的 avro 数据：

avro.io.AvroTypeException: The datum �bankbankHoward �����THoward �����T� is not an example of the schema {
  "namespace": "example.avro",
  "type": "record",
  "connect.name": "TABLE_NAME",
  "fields": [
    {
      "type": "int",
      "name": "Column_1"
    },
    ... (omitting the rest of the schema)

我从here 读到，来自 Confluent 框架的 Avro 格式的 Kafka 消息在消息的开头有 4 个额外字节，指示模式 ID。我试图去掉 Avro 数据的前 4 个字节，然后将其发送到“serializer.to_json()”，但仍然没有运气。

帮助！

【问题讨论】：

标签： python apache-kafka apache-storm avro

【解决方案1】：

我在通过 Storm Kafka spout 读取 kafka 融合数据时遇到了完全相同的问题。这是对我有用的等效 Java 代码。

    ByteBuffer input = ByteBuffer.wrap(data);
    int id = input.getInt();
    int start = input.position() + 1;
    MyAvroObject obj = null;
    try {
        obj  = datum_reader.read(null, DecoderFactory.get().binaryDecoder(input.array(), start, input.limit(), null));

    } catch (IOException e) {
        e.printStackTrace();
    }
    return obj;

getInt() 和 ByteBuffer 上的 position 方法将指针移动到模式 ID 之后。希望这会有所帮助。

【讨论】：