Azure Stream Analytics行转列：如何转换给定格式的设备时序数据

Hey there! Let's figure out how to turn your device's parameter-value rows into columns (a.k.a. pivoting) in Azure Stream Analytics. I've got a couple of straightforward approaches depending on whether you know all your parameter names upfront or need to handle dynamic ones.

1. Pivot with Fixed Parameter Columns (Known parameter Values) #

If you already know all the possible parameter values (like p1 and p2 in your sample data), you can use a combination of CASE statements and aggregation functions to pivot the data. This is the most direct method for static parameter sets.

Example Query #

SELECT
    deviceid,
    timestamp,
    -- Map each parameter to a dedicated column
    MAX(CASE WHEN parameter = 'p1' THEN value END) AS p1,
    MAX(CASE WHEN parameter = 'p2' THEN value END) AS p2
INTO
    [YourOutputSink] -- Replace with your actual output sink (e.g., Azure SQL, Blob Storage)
FROM
    [YourInputStream] -- Replace with your input stream name
GROUP BY
    deviceid,
    timestamp

How It Works #

• We group the data by deviceid and timestamp to ensure each row represents a single device at a specific time.
• The CASE statements filter rows by parameter and pull the corresponding value. Using MAX (or MIN, FIRST—whichever fits your data) ensures we get the single value for that parameter at the grouped time (since each (deviceid, timestamp, parameter) should have one unique value).

For Time-Windowed Aggregates #

If you want to aggregate data over time (e.g., average values per minute), adjust the query to use a tumbling or hopping window:

SELECT
    deviceid,
    System.Timestamp() AS window_end,
    AVG(CAST(CASE WHEN parameter = 'p1' THEN value AS FLOAT)) AS p1_avg,
    MAX(CAST(CASE WHEN parameter = 'p2' THEN value AS FLOAT)) AS p2_max
INTO
    [YourOutputSink]
FROM
    [YourInputStream]
GROUP BY
    deviceid,
    TumblingWindow(minute, 1) -- Aggregate over 1-minute windows

2. Handle Dynamic Parameters (Unknown/Changing parameter Values) #

Azure Stream Analytics doesn't support dynamic pivoting out of the box (since SQL requires explicit column names). But you can use a JavaScript User-Defined Function (UDF) to bundle parameters into a JSON object, which you can later expand in downstream systems (like Power BI or Azure Functions).

Step 1: Create the JavaScript UDF #

1. In your ASA job, go to Functions > Add > JavaScript UDF.
2. Name it CombineParameters and use this code:

function main(events) {
    const result = {
        deviceid: events[0].deviceid,
        timestamp: events[0].timestamp
    };
    // Loop through events to map parameters to their values
    events.forEach(event => {
        result[event.parameter] = event.value;
    });
    return result;
}

Step 2: Use the UDF in Your Query #

SELECT
    CombineParameters(CollectArray(*)) AS pivoted_device_data
INTO
    [YourOutputSink]
FROM
    [YourInputStream]
GROUP BY
    deviceid,
    timestamp

Output Example #

The output will be a JSON object like this for each device-time group:

{
    "deviceid": "d1",
    "timestamp": "2018-03-22T12:34:00",
    "p2": "2"
}

You can then use tools like Power BI's "Expand to New Columns" feature to turn these JSON keys into actual columns.

3. Get Latest Values for All Parameters in One Row #

If you want a single row per device showing the latest value for each parameter (instead of one row per timestamp), use the LATEST window function:

SELECT
    deviceid,
    System.Timestamp() AS last_updated,
    LATEST(CASE WHEN parameter = 'p1' THEN value END) OVER (PARTITION BY deviceid LIMIT DURATION(hour, 1)) AS p1_latest,
    LATEST(CASE WHEN parameter = 'p2' THEN value END) OVER (PARTITION BY deviceid LIMIT DURATION(hour, 1)) AS p2_latest
INTO
    [YourOutputSink]
FROM
    [YourInputStream]

This will keep updating each device's row with the most recent p1 and p2 values from the last hour.

内容的提问来源于stack exchange，提问作者Metehan Mutlu