使用Python通过Kafka以JSON格式收发图像 消费端如何还原NumPy数组

注意：你需要在生产端将NumPy数组的形状（shape）、*数据类型（dtype）*和base64字符串一起存入JSON，否则解码后得到的一维字节流无法还原为原始的多维图像结构。

生产端补充元数据（新增逻辑） #

生产端组装JSON时增加元数据字段，完整处理逻辑如下：

import json
import base64
import numpy as np

# 原有图像处理逻辑
a = some_image_numpy_array # 示例：形状为(720, 1280, 3)、dtype为uint8的图像数组
b = base64.b64encode(a)
c = b.decode('utf-8')

# 组装可序列化为JSON的对象
kafka_msg = json.dumps({
    "img_b64": c,
    "img_shape": a.shape,
    "img_dtype": str(a.dtype)
})
# 后续将kafka_msg传入Kafka pipeline即可

消费端解码逻辑 #

import json
import base64
import numpy as np

# 从Kafka消费到消息后，先解析JSON
received_msg = json.loads(received_kafka_value)
# 取出对应字段
img_b64 = received_msg["img_b64"]
img_shape = tuple(received_msg["img_shape"])
img_dtype = received_msg["img_dtype"]

# 1. base64字符串转回原始字节
img_raw_bytes = base64.b64decode(img_b64)
# 2. 字节流转为一维NumPy数组
img_arr = np.frombuffer(img_raw_bytes, dtype=img_dtype)
# 3. 还原为原始图像形状
img_arr = img_arr.reshape(img_shape)

# 若后续需要修改数组，可改为生成可写副本：
# img_arr = np.frombuffer(img_raw_bytes, dtype=img_dtype).copy()

补充说明 #

• 上述方案适配你当前直接序列化原始NumPy数组的场景，如果你传输的是JPEG/PNG等压缩后的图像数据，解码后需额外调用cv2.imdecode处理。
• 元数据传输开销极低，不会明显增加Kafka消息体积。

内容的提问来源于stack exchange，提问作者sourabh-burnwal