代码之家 › 专栏 › 技术社区 › 0x26res

区分二进制编码的Avro和JSON消息

confluent-schema-registry avro apache-kafka python

0x26res · 技术社区 · 7 年前

我使用python来读取来自不同主题的消息。有些主题的消息是用纯JSON编码的,而另一些主题则使用Avro二进制序列化和合流模式注册表。

当我收到一条信息时,我需要知道它是否需要解码。目前,我只依赖于这样一个事实:二进制编码的消息以 MAGIC_BYTE 哪个值为零:

from confluent_kafka.cimpl import Consumer

consumer = Consumer(config)
consumer.subsrcibe(...)
msg = consumer.poll()
# check the msg is not null or error etc
if msg.values()[0] == 0:
      # It is binary encoded
else:
      # It is json

我想知道有没有更好的办法?

2 回复 | 直到 7 年前

OneCricketeer Gabriele Mariotti 7 年前

你可以得到字节 0-5 你的信息,然后

magic_byte = message_bytes[0]
schema_id = message_bytes[1:5]

然后,对注册表执行查找 GET /schemas/{schema_id} ,并在获得 200 响应代码。

否则,消息要么是JSON,要么是生产者将其数据发送到了另一个注册表(如果您的环境中有多个注册表)。 _{注:这意味着数据可能仍然是Avro}

mjuarez 7 年前

您可以先通过REST查询schema注册表,然后构建在其中注册的主题的本地缓存。然后,当您试图解码来自某个特定主题的消息时,只需将该主题与该列表的内容进行比较。如果它在那里,你知道它已经被解码了。

当然,只有当Avro编码的所有主题都使用Schema Registry时,这才有效。如果你收到一个Avro编码的消息不在架构注册表中注册,则无法工作。

推荐文章

user9750148 · Scala-如何从Avro模式中获取所有字段的名称?

7 年前

user9750148 · 如何从Avro模式创建有效的JSON示例?

7 年前

Suchita · 如何使用Avro反序列化器和Avro序列化器为KafkaStreams编写Junit测试用例

7 年前

CuriousMind · Hadoop自身的序列化及其与AVRO序列化的关系?

7 年前

revy · 配置单元AVRO表创建语法

7 年前

cscan ssice · 反序列化Avro序列化Kafka流时出现问题

7 年前

Atul Ojha · java上使用Avro发送和接收日期类型

7 年前

Pasq · 具有泛型类型问题的Avro序列化

7 年前

Souhaib Guitouni · 将Spark结构化流媒体与融合模式注册表集成

7 年前

M80 · 通过为包含JSON的列定义模式来创建配置单元表的视图

8 年前