如何使用python发布kafka模式

np8igboo 于 2021-06-04 发布在 Kafka

关注(0)|答案(1)|浏览(367)

我正在尝试使用python发布一个kafka模式。
在cli中，我将使用如下语法：

curl -X POST -H "Content-Type: application/vnd.schemaregistry.v1+json" --data '{"schema": "{\"type\":\"record\",\"name\":\"VisualDetections\",\"namespace\":\"com.namespace.something\",\"fields\":[{\"name\":\"vehicle_id\",\"type\":\"int\"},{\"name\":\"source\",\"type\":\"string\"},{\"name\":\"width\",\"type\":\"int\"},{\"name\":\"height\",\"type\":\"int\"},{\"name\":\"annotated_frame\",\"type\":[\"string\",\"null\"]},{\"name\":\"version\",\"type\":\"string\"},{\"name\":\"fps\",\"type\":\"int\"},{\"name\":\"mission_id\",\"type\":\"int\"},{\"name\":\"sequence\",\"type\":{\"type\":\"array\",\"items\":{\"type\":\"record\",\"name\":\"sequence_record\",\"fields\":[{\"name\":\"frame_id\",\"type\":\"int\"},{\"name\":\"timestamp\",\"type\":\"long\"},{\"name\":\"localization\",\"type\":{\"type\":\"array\",\"items\":{\"type\":\"record\",\"name\":\"localization_record\",\"fields\":[{\"name\":\"latitude\",\"type\":\"double\"},{\"name\":\"longitude\",\"type\":\"double\"},{\"name\":\"class\",\"type\":\"string\"},{\"name\":\"object_id\",\"type\":\"int\"},{\"name\":\"confidence\",\"type\":\"double\"},{\"name\":\"bbox\",\"type\":{\"type\":\"record\",\"name\":\"bbox\",\"fields\":[{\"name\":\"x_min\",\"type\":\"int\"},{\"name\":\"y_min\",\"type\":\"int\"},{\"name\":\"x_max\",\"type\":\"int\"},{\"name\":\"y_max\",\"type\":\"int\"}]}}]}}}]}}}]}"}' http://server_ip:8081/subjects/VisualDetections-value/versions/

当我尝试将此函数转换为python时，我尝试了如下操作：

import requests
import json

topic = 'VisualDetections'
headers = {'Content-Type':  'application/vnd.schemaregistry.v1+json'}
with open(avro_path) as fp:
     data = {'schema': json.load(fp)}
data_json = json.dumps(data)
cmd = 'http://server_ip:8081/subjects/{}-value/versions/'.format(topic)
response = requests.post(cmd, headers=headers, data=data_json)

上面返回一个代码 {"error_code":500,"message":"Internal Server Error"} . 我尝试过其他选择，如：

with open(avro_path) as fp:
    data = json.load(fp)

错误代码：

"error_code":422,"message":"Unrecognized field: name"

在上面 avro_path 只在json文件中包含avro模式（如果有用的话也可以上传）。
我不知道如何才能准确地发布这些数据。而且，我没有考虑到 -H 参数，因为我找不到等效的python参数（但不确定它是否起任何作用）。谁能为这个问题提供一个解决方案。

python apache-kafka rest confluent-schema-registry

来源：https://stackoverflow.com/questions/65190118/how-to-post-a-kafka-schema-using-python

1条答案

按热度按时间

vmdwslir1#

对于第二个错误，有效负载需要 {'schema': "schema string"} 首先，我认为这是一个编码的问题； json.load 将文件读入dict而不仅仅是字符串。
通知

>>> import json
>>> schema = {"type":"record"}  # example when using json.load() ... other data excluded
>>> json.dumps({'schema': schema})
'{"schema": {"type": "record"}}'  # the schema value is not a string
>>> json.dumps({'schema': json.dumps(schema)})
'{"schema": "{\\"type\\": \\"record\\"}"}'  # here it is

试着读一下文件

url = 'http://server_ip:8081/subjects/{}-value/versions/'.format(topic)
with open(avro_path) as fp:
     data = {'schema': fp.read().strip()}
     response = requests.post(cmd, headers=headers, data=json.dumps(data))

否则，你会的 json.load 然后使用 json.dumps 两次如上所示
你也可以试试 json=data 而不是 data=json.dumps(data)

赞(0）回复(0）举报 2021-06-04

我来回答

如何使用python发布kafka模式

1条答案

相关问题

热门标签

最新问答