How do I pass all the JSON packets received in that particular tumbling window to the Javascript UDF function from the Azure Stream Analytics Query?

Question

Below is my Azure Stream Analytics query. In this query, I have set the tumbling window for 10 seconds, and whatever data I will be receiving in these 10 seconds, I wanted to pass these complete packets in the "AggrFunction(IotMessage)" UDF function. But we were only receiving one packet/row in this UDF function. I was expecting that the UDF function would receive a collection of objects. However, the UDF is receiving single records amongst the collection of records in the 10-second window.

-- Define the input alias
WITH [AggregatedData] AS (
    SELECT
        System.Timestamp() AS WindowStart,
        UDF.AggrFunction(IotMessage) AS Message
    FROM
        [IoTOutput] AS IotMessage

 

    GROUP BY
        TumblingWindow(second, 10),
        IotMessage

)

 

-- Define the output alias
SELECT
    [Message].*
INTO
    [OEEIn]
FROM
    [AggregatedData]

I was sending the below packets from my end devices with a frequency of 2 seconds:

{      "Temp1": 40,      "Temp2": 50,      "Temp3": 60,      "Temp4": 70,      "Temp5": 70    }

Due to a 10 seconds of tumbling window, I am expecting that I will be receiving the below array of JSON objects in the "AggrFunction()" UDF function from Azure Analytics query.

[
    {
      "Temp1": 40,
      "Temp2": 50,
      "Temp3": 60,
      "Temp4": 70,
      "Temp5": 70
    },
    {
      "Temp1": 10,
      "Temp2": 15,
      "Temp3": 20,
      "Temp4": 30,
      "Temp5": 70
    },
    {
      "Temp1": 25,
      "Temp2": 35,
      "Temp3": 37,
      "Temp4": 48,
      "Temp5": 70
    },
    {
      "Temp1": 58,
      "Temp2": 77,
      "Temp3": 90,
      "Temp4": 71,
      "Temp5": 70
    }
]

Answer

Hello Deepak Chaurasiya,

My understanding is that in Azure Stream Analytics, User-Defined Functions are designed to process each input record individually. So UDF will receive one record at a time, not a collection of records.

This is the reason you are seeing a single record in your UDF.

To process a collection of records within a window, you can use the aggregate function - collect()

collect() Returns an array with all record values from the window.

You can modify your query to collect the records into an array using the collect() and then pass array to your UDF.

Please modify your code below and see if it helps.

-- Define the input alias
WITH [AggregatedData] AS (
    SELECT
        System.Timestamp() AS WindowStart,
        UDF.AggrFunction(collect(IotMessage)) AS Message
    FROM
        [IoTOutput] AS IotMessage

 

    GROUP BY
        TumblingWindow(second, 10),
        IotMessage

)

 

-- Define the output alias
SELECT
    [Message].*
INTO
    [OEEIn]
FROM
    [AggregatedData]

https://video2.skills-academy.com/en-us/stream-analytics-query/collect-azure-stream-analytics

Share via

How do I pass all the JSON packets received in that particular tumbling window to the Javascript UDF function from the Azure Stream Analytics Query?

1 answer