Azure IOT Hub Python SDK application giving error in Stream Analytics "Could not deserialize the input event(s) from resource"

Question

I am trying to send data to IOT Hub and store it in blob storage as well as Power BI streaming dataset using Azure Stream Analytics service.

For sending data, I used Microsoft's Python SDK application (link given below):

https://github.com/Azure/azure-iot-sdk-python/blob/main/samples/async-hub-scenarios/send_message.py

This script successfully sends data to Azure IOT Hub and I am able to visualize it in Azure IOT explorer but does not create a blob storage file or Power BI streaming dataset. Whereas in Stream Analytics, in the input section it gives the below error:

ERROR : "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"

The output I received on Azure IOT explorer is as follows which is a perfect json string:

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

WHEREAS for the same IOT Hub and Stream Analytics connections, if I use the Microsoft's Javascript SDK application (link given below):

https://github.com/Azure/azure-iot-sdk-node/blob/main/device/samples/javascript/simple_sample_device.js

I am able to send data to IOT Hub as well as create a blob storage and streaming dataset on Power BI.

The output received on IOT explorer after running Javascript script.

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

The Javascript application runs absolutely fine but not the python script.

Can anyone please point out what exactly I am doing wrong with Python script?

Answer 1

Hello @Waqar Mehdi,

welcome to this moderated Azure community forum.

based on the error you get, it looks like you do not use separate consumer groups while reading messages from the IoT Hub.

Please use a separate consumer group for each application or service using the event hub compatible endpoint of the IoT Hub (thus not $default)

By using separate consumer groups, each groups gets a copy of the same message:

Next to that, do you also use routing?

If, so read the text about the default event hub compatible endpoint turning in a fall back route:

Make sure you add that extra route for the default message endpoint:

Success!

---

This answer is written by myself, not by some AI. If the response helped, do "Accept Answer". If it doesn't work, please let us know the progress. All community members with similar issues will benefit by doing so. Your contribution is highly appreciated.

Answer 2

Hi @Waqar Mehdi Greetings! Welcome to Microsoft Q&A forum. Thank you for posting this question here.

In addition to what @Sander van de Velde has mentioned, it looks from the error message of Stream Analytics job that the query failed to extract the data from the Input. Assuming that your Analytics job query is to extract deviceId, windSpeed, temperature and humidity details from the input of the stream analytics job, it makes sense that the job fails for Python SDK. The python sample you have shared sends a message to IoT Hub which does not include any of these properties. You need to modify the send_test_message as follows to include these properties in the telemetry.

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

This would include the properties needed by the stream analytics query in the Message telemetry. This should help resolve the issue.

Hope this helps. Please let us know if you have any additional questions.

---

If the response helped, please do click Accept Answer and Yes for the answer provided. Doing so would help other community members with similar issue identify the solution. I highly appreciate your contribution to the community.