Call Azure Stream Analytics UDF with multi-dimensional array of last 5 records, grouped by record

user_2063902 0
Viewed 374

I am trying to call an AzureML UDF from Stream Analytics query and that UDF expects an array of 5 rows and 2 columns. The input data is streamed from an IoT hub and we have two fields in the incoming messages: temperature & humidity.

This would be the 'passthrough query' :

SELECT GetMetadataPropertyValue([room-telemetry], 'IoTHub.ConnectionDeviceId') AS RoomId, 
       Temperature, Humidity
INTO
    [maintenance-alerts]
FROM
    [room-telemetry]

I have an AzureML UDF (successfully created) that should be called with the last 5 records per RoomId and that will return one value from the ML Model. Obviously, there are multiple rooms in my stream, so I need to find a way to get some kind of windowing of 5 records Grouped per RoomId. I don't seem to find a way to call the UDF with the right arrays selected from the input stream. I know I can create a Javascript UDF that would return an array from the specific fields, but that would be record/by record, where here I would need this with multiple records that are grouped by the RoomId.

Someone has any insights?

Best regards

azure-stream-analytics azure-iot-hub azure-machine-learning-service
1 Answers

After the good suggestion of @jean-sébastien and an answer to an isolated question for the array-parsing, I finally was able to stitch everything together in a solution that builds. (still have to get it to run at runtime, though).

So, the solution exists in using CollectTop to aggregate the latest rows of the entity you want to group by, including the specification of a Time Window.

And the next step was to create the javascript UDF to take that data structure and parse it into a multi-dimensional array.

This is the query I have right now:

-- Taking relevant fields from the input stream
WITH RelevantTelemetry AS
(
    SELECT  engineid, tmp, hum, eventtime
    FROM    [engine-telemetry] 
    WHERE   engineid IS NOT NULL
),
-- Grouping by engineid in TimeWindows
TimeWindows AS
(
    SELECT engineid, 
        CollectTop(2) OVER (ORDER BY eventtime DESC) as TimeWindow
    FROM
        [RelevantTelemetry]
    WHERE engineid IS NOT NULL
    GROUP BY TumblingWindow(hour, 24), engineid
)
--Output timewindows for verification purposes
SELECT engineid, Udf.Predict(Udf.getTimeWindows(TimeWindow)) as Prediction
INTO debug
FROM TimeWindows

And this is the Javascript UDF:

    function getTimeWindows(input){
        var output = [];
        for(var x in input){
            var array = [];
            array.push(input[x].value.tmp);
            array.push(input[x].value.hum);
            output.push(array);
        }
        return output;
    }
user_20639020
Related