Azure IOT Hub Python SDK 应用程序在流分析中给出错误“无法从资源反序列化输入事件”

Question

我正在尝试使用 Azure 流分析服务将数据发送到 IOT 中心并将其存储在 Blob 存储以及 Power BI 流数据集中。

为了发送数据，我使用了 Microsoft 的 Python SDK 应用程序（下面给出的链接）：

Python应用代码链接

此脚本成功将数据发送到 Azure IOT 中心，我可以在 Azure IOT 资源管理器中对其进行可视化，但不会创建 blob 存储文件或 Power BI 流数据集。而在流分析中，在输入部分它会给出以下错误：

错误：“无法将资源“分区：[1]，偏移量：[8591414904]，序列号：[3081]，DeviceId：[test001]”中的输入事件反序列化为 Json。一些可能的原因：1）格式错误事件 2) 输入源配置了不正确的序列化格式”

我在Azure IOT explorer上收到的输出如下，这是一个完美的json字符串：

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

对于相同的 IOT 中心和流分析连接，如果我使用 Microsoft 的 Javascript SDK 应用程序（下面给出的链接）：

Javascript应用程序代码链接

我能够将数据发送到 IOT Hub，并在 Power BI 上创建 Blob 存储和流数据集。

运行 Javascript 脚本后在 IOT Explorer 上收到的输出。

{
  "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"
  }
}

Javascript 应用程序运行得绝对正常，但 python 脚本则不行。

任何人都可以指出我的Python脚本到底做错了什么吗？

我尝试运行 Python 以及 Javascript Azure SDK，但只从 Javascript SDK 获得所需的输出，而不是 python 脚本。

Answer 1

欢迎来到 Stack Overflow 社区！从流分析作业的错误消息来看，查询在尝试从输入中提取/处理数据时出错。假设您的 Analytics 作业查询旨在从输入中提取 deviceId、windSpeed、温度和湿度详细信息，则 Python SDK 作业失败是有道理的。您共享的 python 示例向 IoT 中心发送一条消息，该消息不包含任何这些属性。您需要按如下方式修改 send_test_message 以将这些属性包含在遥测中。

async def send_test_message(i):        
    windSpeed = 10 + (random.uniform(0, 4))
    temperature = 20 + (random.uniform(0, 10))
    humidity = 60 + (random.uniform(0, 20))
    msg = Message("deviceId: " + 'myFirstDevice' + " windSpeed: "+ str(windSpeed)+
                   " temperature:"  +str(temperature)+ " humidity:" +str(humidity))
    print("sending message " + str(msg))
    msg.message_id = uuid.uuid4()
    msg.correlation_id = "correlation-1234"
    msg.custom_properties["tornado-warning"] = "yes"
    msg.content_encoding = "utf-8"
    msg.content_type = "application/json"
    await device_client.send_message(msg)
    print("done sending message #" + str(i))

这将包括消息遥测中流分析查询所需的属性。这应该有助于解决问题。