如何使用 python post kafka 模式

Question

我正在尝试 post 使用 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)

以上returns一个代码{"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 模式（如果有用也可以上传）。

我不确定如何 post 这个数据。此外，我没有考虑 CLI 中 post 的 -H 参数，因为我找不到等效的 python 参数（虽然不确定它是否起作用）。任何人都可以提供解决此问题的方法。

Answer 1

对于第二个错误，payload需要{'schema': "schema string"}

首先，我认为是编码问题； json.load 会将文件读入字典，而不仅仅是字符串。

通知

>>> 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)

如何使用 python post kafka 模式

How to post a kafka schema using python

python

rest

apache-kafka

confluent-schema-registry