如何通过Azure Stream Analytics查询实现IoT传感器数据去重插入Azure SQL？

我来帮你搞定这个重复插入数据的麻烦！你遇到的核心问题是传感器采样时ASA生成了大量重复消息，需要基于Date时间戳（还要结合DeviceId，避免不同设备的同时间数据被误去重）只保留一行数据。

先说说你之前的查询为什么没数据：

• CollectTop(1) OVER (...) as Date会返回一个数组类型的结果，但你的SQL表Date字段是单个时间值，类型不匹配导致无法插入；
• TumblingWindow(second, 60)的窗口粒度可能和你的Date时间戳粒度不匹配（比如你的Date是秒级甚至毫秒级），导致没有数据被匹配到窗口里。

下面给你两种可靠的解决方案：

方案1：用ROW_NUMBER()窗口函数精准去重 #

这种方法适合需要保留每组第一条数据（按入队时间排序）的场景，能确保每个DeviceId+Date组合只存一行：

WITH DeduplicatedData AS (
    SELECT 
        *,
        -- 按设备ID和时间戳分组，给每组的行编号，入队最早的行编号为1
        ROW_NUMBER() OVER (PARTITION BY DeviceId, Date ORDER BY EventEnqueuedUtcTime ASC) AS RowNum
    FROM mystreamin
    TIMESTAMP BY Time -- 这里的Time要和你的Date字段对应，确保是事件的实际时间
)
SELECT 
    Date, DeviceId, Temperature, Humidity, Moisture, 
    EventProcessedUtcTime, PartitionId, EventEnqueuedUtcTime, IoTHub, EventID
INTO myazuredb 
FROM DeduplicatedData
WHERE RowNum = 1 -- 只保留每组的第一行

方案2：用GROUP BY+聚合函数去重 #

如果同一DeviceId+Date下的所有字段值都完全相同，你可以用聚合函数（MAX/MIN）来提取唯一值，这种写法更简洁：

SELECT 
    Date, 
    DeviceId,
    MAX(Temperature) AS Temperature, -- 因为重复数据值相同，MAX/MIN结果一致
    MAX(Humidity) AS Humidity,
    MAX(Moisture) AS Moisture,
    MIN(EventEnqueuedUtcTime) AS EventEnqueuedUtcTime, -- 取最早的入队时间
    MAX(EventProcessedUtcTime) AS EventProcessedUtcTime,
    MAX(PartitionId) AS PartitionId,
    MAX(IoTHub) AS IoTHub,
    MAX(EventID) AS EventID
INTO myazuredb
FROM mystreamin
TIMESTAMP BY Time
-- 窗口粒度要匹配Date的精度，如果Date是秒级就用1秒，毫秒级就用1毫秒
GROUP BY DeviceId, Date, TumblingWindow(second, 1)

额外注意事项 #

1. 确认TIMESTAMP BY的Time字段是正确的事件时间，要和Date字段的时间精度一致，避免窗口分组错位；
2. 调试时可以先把查询结果输出到Azure Blob Storage，查看去重后的数据是否符合预期，再切换到SQL数据库；
3. 如果你的Date字段包含毫秒，记得把窗口调整为TumblingWindow(millisecond, 1)。

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