如何以编程方式从 Python 中的融合模式注册表中获取模式

Posted 2023-03-10

【中文标题】如何以编程方式从 Python 中的融合模式注册表中获取模式

【英文标题】：How to programmatically get schema from confluent schema registry in Python

【发布时间】：2020-06-13 12:16:26

【问题描述】：

到目前为止，我正在做类似这样的读取 avsc 文件来获取架构

value_schema = avro.load('client.avsc')

我可以做些什么来使用主题名称从融合模式注册表中获取模式吗？

我找到了一种方法，但不知道如何使用它。

https://github.com/marcosschroh/python-schema-registry-client

【问题讨论】：

您只需要一个 HTTP 客户端即可访问 REST API...

【参考方案1】：

使用confluent-kafka-python

from confluent_kafka.avro.cached_schema_registry_client import CachedSchemaRegistryClient

sr = CachedSchemaRegistryClient(
    'url': 'http://localhost:8081',
    'ssl.certificate.location': '/path/to/cert',  # optional
    'ssl.key.location': '/path/to/key'  # optional
)

value_schema = sr.get_latest_schema("orders-value")[1]
key_schema= sr.get_latest_schema("orders-key")[1]

---

使用SchemaRegistryClient

通过主题名称获取架构

from schema_registry.client import SchemaRegistryClient


sr = SchemaRegistryClient('localhost:8081')
my_schema = sr.get_schema(subject='mySubject', version='latest')

通过 ID 获取架构

from schema_registry.client import SchemaRegistryClient


sr = SchemaRegistryClient('localhost:8081')
my_schema = sr.get_by_id(schema_id=1)

【讨论】：

没错。只是要添加一个注释，通常主题的主题将是 -key 或 -value ，具体取决于您正在阅读的消息的哪一部分。此外，您可以改为（按照我的理解是融合 serde）以某种方式将实际模式 id 编码到序列化消息中，然后通过它而不是主题和版本访问模式。

@wobr 其实你可以省略version。在这种情况下，您将获得最新的主题架构。

最新的并不总是你想要的。

@wobr confluent_kafka Python 库以某种方式处理使用 ID 编码消息的“某种方式”

@wobr - 所以你的意思是说对于每条记录，从模式注册表中获取模式？会不会拖慢性能？

【参考方案2】：

您可以使用get_latest_version 函数获取架构信息

from confluent_kafka.schema_registry import SchemaRegistryClient

sr = SchemaRegistryClient("url": 'http://localhost:8081')
subjects = sr.get_subjects()
for subject in subjects:
    schema = sr.get_latest_version(subject)
    print(schema.version)
    print(schema.schema_id)
    print(schema.schema.schema_str)

【参考方案3】：

我确实喜欢它对我有用

     import requests
     import os

     SCHEMA_REGISTRY_URL = os.getenv('SCHEMA_REGISTRY_URL');
     print("SCHEMA_REGISTRY_URL: ", SCHEMA_REGISTRY_URL)
     URL = SCHEMA_REGISTRY_URL + '/subjects/' + topic + '/versions/latest/schema'
     r = requests.get(url=URL)
     schema = r.json()


     print("Schema From Schema Registry ==========================>>")
     print("Schema: ", schema)