Azure 流分析中的异常情况检测

在云和Azure IoT Edge中都可用，Azure 流分析提供了基于机器学习的内置异常检测功能，可用于监视两种最常见的异常：临时和持久性。 通过使用 AnomalyDetection_SpikeAndDip 和 AnomalyDetection_ChangePoint 函数，可以直接在流分析作业中执行异常情况检测。

机器学习模型假定采用统一采样的时序。 如果时序不统一，在调用异常检测之前，请使用翻滚窗口插入聚合步骤。

机器学习作目前不支持季节性趋势或多变量相关性。

模范行为

通常，模型的准确性会随着滑动窗口中的更多数据而提高。 指定滑动窗口中的数据被视为该时间范围内其正常值范围的一部分。 该模型仅考虑滑动窗口中的事件历史记录，以检查当前事件是否异常。 当滑动窗口移动时，旧的数据将从模型的训练中被移除。

这些函数通过根据到目前为止所看到的情况建立一定的正常标准来运行。 通过在置信度级别内根据建立的法线进行比较来识别离群值。 窗口大小应基于为正常行为训练模型所需的最小事件，以便在发生异常时能够识别它。

模型的响应时间随历史记录大小而增加，因为它需要与更多的过去事件进行比较。 为了获得更好的性能，请仅包含必要的事件数。

当模型未在特定时间点接收事件时，时序中的差距可能会发生。 流分析通过使用插补逻辑来处理这种情况。 历史记录大小以及同一滑动窗口的持续时间用于计算预期到达事件的平均速率。

可以使用 anomaly 生成器向 IoT 中心 提供包含不同异常模式的数据。 可以通过使用这些异常检测函数，从此IoT 中心读取数据并检测异常，以设置Azure 流分析作业。

峰值和下降

时序事件流中的临时异常称为峰值和下降。 可以使用基于机器学习的运算符AnomalyDetection_SpikeAndDip监视峰值和下降。

在同一滑动窗口中，如果第二个峰值小于第一个峰值，则与指定置信度内的第一个峰值的分数相比，较小的峰值的计算分数可能不够重要。 可以尝试降低模型的置信度来检测此类异常。 但是，如果开始收到过多的警报，请使用更高的置信区间。

以下示例查询假定在包含120个事件历史记录的2分钟滑动窗口中，每秒输入一个事件，以统一的速度进行。 最终的 SELECT 语句提取并输出分数和异常状态，置信度为 95%。

WITH AnomalyDetectionStep AS
(
    SELECT
        EVENTENQUEUEDUTCTIME AS time,
        CAST(temperature AS float) AS temp,
        AnomalyDetection_SpikeAndDip(CAST(temperature AS float), 95, 120, 'spikesanddips')
            OVER(LIMIT DURATION(second, 120)) AS SpikeAndDipScores
    FROM input
)
SELECT
    time,
    temp,
    CAST(GetRecordPropertyValue(SpikeAndDipScores, 'Score') AS float) AS
    SpikeAndDipScore,
    CAST(GetRecordPropertyValue(SpikeAndDipScores, 'IsAnomaly') AS bigint) AS
    IsSpikeAndDipAnomaly
INTO output
FROM AnomalyDetectionStep

变化点

时序事件流中的持久异常是事件流中值分布的变化，例如级别更改和趋势。 在流分析中，基于 AnomalyDetection_ChangePoint 运算符机器学习检测这些异常。

持久更改的持续时间比峰值和下降要长得多，并可能表示灾难性事件。 永久性更改通常对裸眼不可见，但 AnomalyDetection_ChangePoint 运算符可以检测到它们。

下图是级别更改的示例：

下图是趋势更改的示例：

以下示例查询假定在 20 分钟的滑动窗口中，输入速率为每秒一个事件，且历史记录大小为 1,200 个事件。 最终的 SELECT 语句提取并输出分数和异常状态，置信度为 80%。

WITH AnomalyDetectionStep AS
(
    SELECT
        EVENTENQUEUEDUTCTIME AS time,
        CAST(temperature AS float) AS temp,
        AnomalyDetection_ChangePoint(CAST(temperature AS float), 80, 1200) 
        OVER(LIMIT DURATION(minute, 20)) AS ChangePointScores
    FROM input
)
SELECT
    time,
    temp,
    CAST(GetRecordPropertyValue(ChangePointScores, 'Score') AS float) AS
    ChangePointScore,
    CAST(GetRecordPropertyValue(ChangePointScores, 'IsAnomaly') AS bigint) AS
    IsChangePointAnomaly
INTO output
FROM AnomalyDetectionStep

性能特征

这些模型的性能取决于历史记录大小、窗口持续时间、事件加载以及是否使用函数级别分区。 本部分讨论了这些配置，并提供了有关如何维持每秒 1 K、5 K 和 10 K 事件的引入速率的示例。

• 历史记录大小 - 这些模型使用 历史记录大小以线性方式执行。 历史记录大小越长，模型评分新事件所花费的时间就越长。 模型将新事件与历史记录缓冲区中的每个过去事件进行比较。
• 窗口持续时间 - 窗口持续时间 应反映接收历史记录大小指定的事件所需的时间。 如果窗口中没有这么多事件，Azure 流分析会插补缺失值。 因此，CPU 消耗量是历史记录大小的函数。
• 事件负载 - 事件负载越大，模型执行的工作量就越大，这会影响 CPU 消耗。 可以通过使作业变得尴尬地并行进行横向扩展，假设业务逻辑使用更多输入分区是有意义的。
• 函数级别分区 - 在 PARTITION BY 异常情况检测函数调用中使用来执行 函数级分区。 这种类型的分区会增加开销，因为作业需要同时维护多个模型的状态。 在设备级别分区等方案中使用函数级分区。

关系

历史记录大小、窗口持续时间和事件总负载按以下方式相关：

windowDuration (毫秒) = 1000 * historySize / (每秒输入事件总数 / 输入分区计数)

按 deviceId 对函数进行分区时，将“PARTITION BY deviceId”添加到异常检测函数调用中。

观测结果

下表显示了非分区事例的单个节点（6 SU）的吞吐量观察值：

下表显示了分区事例的单个节点（6 SU）的吞吐量观察值：

可以在Azure示例的 Streaming At Scale 存储库中找到运行非分区配置的示例代码。 该代码创建一个没有函数级分区的流分析作业，该作业使用事件中心作为输入和输出。 测试客户端生成输入负载。 每个输入事件都是一个 1 KB JSON 文档。 这些事件模拟 IoT 设备发送 JSON 数据（最多 1 K 台设备）。 历史记录大小、窗口持续时间和事件总负载因两个输入分区而异。

识别瓶颈

若要确定管道中的瓶颈，请使用Azure 流分析作业中的“指标”窗格。 查看 输入/输出事件 ，了解吞吐量和 “水印延迟” 或 积压事件 ，以查看作业是否跟上输入速率。 对于事件中心指标，请查看“限制的请求数”并相应地调整阈值单位。 对于 Azure Cosmos DB 指标，请查看“吞吐量”下的“每个分区键范围所使用的最大 RU/秒”，以确保均匀使用分区键范围。 对于 Azure SQL DB，请监视“日志 IO”和“CPU”。

后续步骤

• Azure 流分析简介
• Azure 流分析入门
• 调整 Azure 流分析作业的规模
• Azure 流分析查询语言参考
• Azure 流分析管理 REST API 参考