Split message stream by device and process in batches

Question

We use the Azure IotHub to receive messages from a couple thousands of sensor devices. Due to constraints on the sensor side the data is sent in small packets in a binary format to an iothub. The devices send e.g. a couple of hundred messages in a few minutes, then there is silence for some time, then they send messages again based on some activity on device sid

On server side we want to collect these packet messages by device and and create a batch, e.g. (min. 500 messages, max. 5 minutes) of data per device and then store this batch either in an storage endpoint or forward the batch to a queue.

Currently, we use a dedicated service process we wrote to consume all messages from the built-in endpoint, stores them in-memory until one of the criteria from above is met and then writes the output file per device.

I wonder if there is a built-in way which would scale better and would be easier to handle.

• My favourite way would be to just use the builtin route to azure storage. However, it is not possible to split the output by device there.
• I tried to use a Azure stream analytics jobs to perform this but could not configure the input correctly since the messages are just binary blobs. Then i wrote a custom deserializer to convert the binary data to base64. We tried to use the storage output and split the messages based on device Id in the path pattern. However this seems to be very inefficient and costly. We needed many streaming units to cope with the incoming messages in this simple pass-through scenario

Do you have any other ideas or are there any available solutions for this scenario since i would suppose this to be a quite common scenario.

Thanks!

Answer 1

Hello @Georg Sieber,

Your scenario is great use-case and indeed a helpful post for many users who want to do benchmarking with various solutions!!!! I really like the way you have tried many different solutions!!! this is a helpful post on this forum!!!

Currently, we use a dedicated service process we wrote to consume all messages from the built-in endpoint, stores them in-memory until one of the criteria from above is met and then writes the output file per device.

It may not be the most scalable or fault-tolerant approach, especially as the number of devices and messages increases in future!!!

One alternative approach you could consider is to use Azure Stream Analytics (ASA) as suggested in the previous response by Matthijs (MVP) and i see that you have already tried that route!

My favourite way would be to just use the builtin route to azure storage. However, it is not possible to split the output by device there.

You are correct that the built-in route to Azure Storage in Azure IoT Hub doesn't support splitting the output by device. One approach you could take is to use Azure Functions

I tried to use a Azure stream analytics jobs to perform this but could not configure the input correctly since the messages are just binary blobs.

It's true ASA require some additional configuration and setup, with binary blobs as input data!!!!

One alternative approach you could consider is to use Azure Functions to process the messages from the IoT Hub and write them to your storage endpoint in batches.

Do you have any other ideas or are there any available solutions for this scenario since i would suppose this to be a quite common scenario.

You can try the Azure Functions, serverless and highly scalable, as Azure Functions automatically scales to meet demand and only charges for the compute resources used during each execution.

You can use an IoT Hub trigger to automatically invoke an Azure Function whenever a message is received by the IoT Hub. Inside the Azure Function, you can then use the device ID to group the incoming messages by device and create a batch and you can forward the batch to your desired destination (e.g. storage endpoint or queue).

a small sample snippet of the Azure functions may look like below...!

        foreach (EventData eventData in events)
        {
            // Extract the device ID from the incoming message
            string deviceId = eventData.SystemProperties["iothub-connection-device-id"].ToString();

            // Add the message to the device's message cache
            if (!deviceMessageCache.ContainsKey(deviceId))
            {
                deviceMessageCache[deviceId] = new List<EventData>();
            }
            deviceMessageCache[deviceId].Add(eventData);

            // Check if the device's message cache meets the batch criteria
            if (deviceMessageCache[deviceId].Count >= 500 || (DateTime.UtcNow - eventData.SystemProperties.EnqueuedTimeUtc).TotalMinutes >= 5)
            {
                // Create a batch and forward it to the desired destination
                List<EventData> messageBatch = deviceMessageCache[deviceId];
                deviceMessageCache[deviceId] = new List<EventData>();
                await ForwardMessageBatchAsync(deviceId, messageBatch);
            }
        }


 private static async Task ForwardMessageBatchAsync(string deviceId, List<EventData> messageBatch)
    {
        // Forward the message batch to the desired destination (e.g. storage endpoint or queue)
        // ...
    }

---

If this answers your query, do click Accept Answer and Yes for this answer as helpful. And, if you have any further query do let us know by commenting in the below section.

Answer 2

You could look into message enrichments. Currently, only the IoT Hub name or fields from the tags or desired/reported properties can be added to a message, but as a workaround, you could add the device ID to IoT Hub as a tag for every device. Every message will then have a tag available on them, and you wouldn't have to deserialize anything in ASA. This might reduce your streaming units to an acceptable level. It might still be more costly than your custom solution.