阿夫罗。io。AvroTypeException：数据不是模式{…}的示例

Question

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

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

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

我们正在使用两个名为“Avro\u json\u 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)

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

救命！

Answer 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;

ByteBuffer上的getInt（）和position方法将指针移动到架构Id之后。希望这有帮助。