Azure Stream Analytics SQL查询：移除Event Hub数据异常值

针对你的需求——实时从Event Hub接收数据，按Id分组并剔除过去60秒内偏差过大的异常值，再将有效数据发送至Azure Functions，以下是具体的实现方案：

核心思路 #

使用**滑动窗口（Sliding Window）**捕获过去60秒的所有数据，按id分组后通过统计指标（均值+标准差或中位数+绝对偏差）界定正常数据范围，最终过滤掉超出范围的异常记录。

示例数据 #

id  Temp    date        datetime
123 30      2023-01-01  2023-01-01 12:00:00
124 35      2023-01-01  2023-01-01 12:00:00
123 31      2023-01-01  2023-01-01 12:00:00
123 33      2023-01-01  2023-01-01 12:00:00
123 60      2023-01-01  2023-01-01 12:00:00
124 36      2023-01-01  2023-01-01 12:00:00
124 36      2023-01-01  2023-01-01 12:00:00
124 8       2023-01-01  2023-01-01 12:00:00
124 36      2023-01-01  2023-01-01 12:00:00

方案1：均值+标准差（适合正态分布数据） #

这种方法适合数据分布相对集中的场景，通过均值±N倍标准差界定正常范围：

WITH GroupStats AS (
    SELECT
        id,
        AVG(Temp) AS AvgTemp,
        STDEV(Temp) AS StdTemp
    FROM
        [YourEventHubInput] TIMESTAMP BY datetime
    GROUP BY
        id,
        SlidingWindow(second, 60)
)

SELECT
    original.id,
    original.Temp,
    original.date,
    original.datetime
FROM
    [YourEventHubInput] original TIMESTAMP BY datetime
INNER JOIN
    GroupStats stats
ON
    original.id = stats.id
    AND DATEDIFF(second, original.datetime, System.Timestamp()) BETWEEN 0 AND 60
WHERE
    -- 可根据业务需求调整标准差倍数，示例用2倍（覆盖约95%的正态分布数据）
    original.Temp BETWEEN stats.AvgTemp - 2 * stats.StdTemp AND stats.AvgTemp + 2 * stats.StdTemp

关键部分说明 #

• TIMESTAMP BY datetime：指定事件时间列，确保窗口计算基于数据的实际生成时间，而非到达时间
• SlidingWindow(second, 60)：持续维护过去60秒的数据集，每有新数据进入就更新窗口
• GroupStats 公共表表达式：预计算每个id分组的温度均值和标准差，作为异常判断基准
• 关联条件：确保原始数据与统计数据属于同一60秒窗口内的同一分组

方案2：中位数+绝对偏差（抗极端值更稳健） #

如果分组内存在大量极端值（比如你的示例中Id123的60、Id124的8），中位数+中位数绝对偏差（MAD）的方法更稳健，不受极端值影响：

WITH GroupStats AS (
    SELECT
        id,
        -- 计算分组中位数
        PERCENTILE_CONT(0.5) WITHIN GROUP (ORDER BY Temp) OVER (PARTITION BY id) AS MedianTemp,
        -- 计算中位数绝对偏差
        AVG(ABS(Temp - PERCENTILE_CONT(0.5) WITHIN GROUP (ORDER BY Temp) OVER (PARTITION BY id))) AS MAD
    FROM
        [YourEventHubInput] TIMESTAMP BY datetime
    GROUP BY
        id,
        SlidingWindow(second, 60)
)

SELECT
    original.id,
    original.Temp,
    original.date,
    original.datetime
FROM
    [YourEventHubInput] original TIMESTAMP BY datetime
INNER JOIN
    GroupStats stats
ON
    original.id = stats.id
    AND DATEDIFF(second, original.datetime, System.Timestamp()) BETWEEN 0 AND 60
WHERE
    -- 1.4826是将MAD转换为标准差等价量的系数，对应正态分布的±2σ范围
    original.Temp BETWEEN stats.MedianTemp - 1.4826 * stats.MAD AND stats.MedianTemp + 1.4826 * stats.MAD

注意事项 #

• 将[YourEventHubInput]替换为你在ASA中配置的Event Hub输入别名
• 可根据业务场景调整窗口时长（比如30秒、120秒）和异常判断的倍数
• 若需输出到Azure Functions，直接将查询结果指向对应的Functions输出即可

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