代码之家 › 专栏 › 技术社区 › Abhimanyu

从python Kafka库中推送Kafka主题数据时发生反序列化错误

kafka-producer-api apache-kafka apache java

0

Abhimanyu · 技术社区 · 7 年前

AVRO模式:

{
    "namespace": "example.avro",
    "type": "record",
    "name": "topicname",
    "fields": [
        {"name": "deviceid", "type": "string"},
        {"name": "longitude", "type": "float"},
        {"name": "latitude",  "type": "float"}
    ]
}

我发布数据的Python代码是:

# Path to user.avsc avro schema
SCHEMA_PATH = "user.avsc"
SCHEMA = avro.schema.parse(open(SCHEMA_PATH).read())

writer = DatumWriter(SCHEMA)
bytes_writer = io.BytesIO()
encoder = avro.io.BinaryEncoder(bytes_writer)
writer.write({"deviceid":"9098", "latitude":  90.34 , "longitude": 334.4}, encoder)
raw_bytes = bytes_writer.getvalue()
PRODUCER.send_messages(TOPIC, raw_bytes)

我在Kafka Connect日志中发现以下错误:

反序列化id为-1的Avro消息\n由以下人员使用: org.apache.kafka.common.errors.SerializationException:未知魔法字节\n,“id”:0,“worker_id”:“0.0.0.0:8083”}],“type”:“sink”}

有什么问题吗?

1 回复 | 直到 5 年前

1

SourceSimian 7 年前

我对各种python客户机没有做太多的工作,但几乎可以肯定的是,这个神奇的字节错误是因为您发送的可能是有效的avro,但是如果您想与模式注册表集成,负载需要采用不同的格式(附加的头信息,在这里有文档记录) https://docs.confluent.io/current/schema-registry/docs/serializer-formatter.html 搜索导线格式或魔法字节)。我个人会尝试使用confluent的python kafka客户端-- https://github.com/confluentinc/confluent-kafka-python --它有使用Avro和模式注册表的示例。