Azure IoT Hub Python SDK数据在Stream Analytics反序列化失败求助

问题现象 #

通过Azure Stream Analytics将IoT Hub数据流转到Blob Storage和Power BI流式数据集时，使用Python SDK示例代码发送的数据能在Azure IoT Explorer正常查看，但Stream Analytics输入模块报错：

Could not deserialize the input event(s) from resource 'Partition: [1], Offset: [8591414904], SequenceNumber: [3081], DeviceId: [test001]' as Json. Some possible reasons: 1) Malformed events 2) Input source configured with incorrect serialization format

同时无法生成Blob存储文件和Power BI流式数据集。

数据格式对比 #

• Python SDK发送的消息body为字符串类型：

{
  "body": "test wind speed 10",
  "enqueuedTime": "Tue Nov 07 2023 23:49:29 GMT+0530 (India Standard Time)",
  "properties": {
    "tornado-warning": "yes"
  }
}

• JavaScript SDK发送的消息body为JSON对象类型，此时Stream Analytics功能正常：

{
  "body": {
    "deviceId": "myFirstDevice",
    "windSpeed": 10.487688712678684,
    "temperature": 23.702052411822482,
    "humidity": 79.27664927491016
  },
  "enqueuedTime": "Tue Nov 07 2023 23:02:30 GMT+0530 (India Standard Time)",
  "properties": {
    "temperatureAlert": "false"
  }
}

问题根源与解决方案 #

根源 #

Stream Analytics输入配置为解析JSON格式数据，但Python SDK示例发送的消息body是纯字符串，而非JSON结构化数据，导致Stream Analytics无法将字符串反序列化为JSON对象，触发解析错误。

修复代码 #

修改Python代码，将JSON对象序列化后作为消息体发送，替换原有的纯字符串消息：

import json
from azure.iot.device import Message

# 构造JSON格式的消息内容
message_payload = {
    "deviceId": "test001",
    "windSpeed": 10,
    # 可根据需求添加更多字段
}

# 将JSON对象序列化为UTF-8编码的字节流
message = Message(json.dumps(message_payload).encode("utf-8"))
# 保留原有属性设置（如果需要）
message.custom_properties["tornado-warning"] = "yes"

# 后续发送逻辑保持不变

验证 #

修改后，IoT Explorer中查看的消息body会变为JSON对象格式，与JavaScript SDK发送的结构一致，Stream Analytics即可正常解析数据，完成向Blob Storage和Power BI的流转。

内容的提问来源于stack exchange，提问作者waqar mehdi