我在Python2.7中使用Avro Producer。我需要发送一条带有密钥和值的消息, 该值在主题中有Avro模式,但该键没有Avro模式(由于键遗留原因,我无法添加模式)
这是我的代码:
def main():
kafkaBrokers = os.environ.get('KAFKA_BROKERS')
schemaRegistry = os.environ.get('SCHEMA_REGISTRY')
topic = os.environ.get('KAFKA_TOPIC')
subject = '${}-value'.format(topic)
sr = CachedSchemaRegistryClient(schemaRegistry)
schema = sr.get_latest_schema(subject).schema
value_schema = avro.loads(str(schema))
url = 'test.com'
value = {'url': u'test.com', 'priority': 10}
avroProducer = AvroProducer({
'bootstrap.servers': kafkaBrokers,
'schema.registry.url': schemaRegistry
}, default_value_schema=value_schema)
key = 1638895406382020875
avroProducer.produce(topic=topic, value=value, key=key)
avroProducer.flush()
我得到以下错误:
raise KeySerializerError("Avro schema required for key")
confluent_kafka.avro.serializer.KeySerializerError: Avro schema required for key
如果我从生成函数中删除密钥:
avroProducer.produce(topic=topic, value=value)
它起作用了
如何在没有模式的情况下发送密钥
AvroProducer
假设键和值都是用模式注册表编码的,在键和值的有效负载前面加上一个魔术字节和模式id如果要对密钥使用自定义序列化,可以使用
Producer
而不是AvroProducer
。但是序列化键(使用您想要的任何格式)和值(这意味着对值进行编码并在magic字节和模式id之前加上前缀)是您的责任。要了解这是如何做到的,您可以查看AvroProducer
代码但这也意味着您必须编写自己的
AvroConsumer
,并且不能使用kafka-avro-console-consumer
您需要使用常规生产者并自己执行序列化函数
相关问题 更多 >
编程相关推荐