ОбнаружениеАномалий_ТочкаИзменения

✅ Поток событий Azure Stream Analytics ✅ Fabric

Обнаруживает постоянные аномалии в потоке событий временных рядов. Базовая модель машинного обучения использует алгоритм Exchangeability Martingales.

Синтаксис

AnomalyDetection_ChangePoint(
        <scalar_expression>, 
        <confidence>, 
        <historySize>)
    OVER ([PARTITION BY <partition key>] 
        LIMIT DURATION(<unit>, <length>)
    [WHEN boolean_expression])

Аргументы

scalar_expression

Столбец событий или вычисляемое поле, с помощью которого модель выполняет обнаружение аномалий. Допустимые значения для этого параметра включают типы данных FLOAT или BIGINT, возвращающие одно (скалярное) значение.

Выражение подстановочного знака * запрещено. Кроме того, scalar_expression не может содержать другие аналитические функции или внешние функции.

уверенность

Процентное число от 1,00 до 100 (включительно), которое задает чувствительность модели машинного обучения. Чем ниже достоверность, тем выше число обнаруженных аномалий и наоборот. Начните с произвольного числа от 70 до 90 и измените это на основе результатов, наблюдаемых в разработке или тестировании.

historySize

Количество событий в скользящем окне, которое модель постоянно изучает и использует для оценки следующего события для аномальной активности. Как правило, это должно представлять период времени нормального поведения, чтобы модель помечала последующие аномалии. Начните с образованных предположений с использованием исторических журналов и корректируйте на основе результатов, наблюдаемых в разработке или тестировании.

OVER ([ partition_by_clause ] limit_duration_clause [when_clause])

partition_by_clause

Используется для секционирования обучения модели на основе определенного столбца в событиях. Модель применяет одни и те же параметры параметров функции во всех секциях.

limit_duration_clause DURATION(unit, length)

Размер скользящего окна в Stream Analytics с точки зрения времени. Рекомендуемый размер этого периода времени эквивалентен времени, затрачиваемого на создание числа событий журнала в устойчивом состоянии.

when_clause

Указывает логическое условие для событий, предоставляемых модели для обнаружения аномалий. When_clause является необязательным.

Типы возвращаемых данных

Функция возвращает вложенную запись, состоящую из следующих столбцов:

IsAnomaly

BigINT (0 или 1), указывающий, было ли событие аномальным или нет.

Балл

Вычисляемая оценка Martingale (float), указывающая, как аномально событие. Эта оценка увеличивается экспоненциально с аномальными значениями.

Примеры

В следующем примере запроса первый запрос предполагает событие каждые 5 минут, а второй запрос предполагает событие каждую секунду. Уровень достоверности устанавливается на уровне 75 для обеих моделей.

AnomalyDetection_ChangePoint(reading, 75, 72)
    OVER (LIMIT DURATION(hour, 6))

AnomalyDetection_ChangePoint(temperature, 75, 120)
    OVER ([PARTITION BY sensorId] LIMIT DURATION(second, 120))

Предположим, что в 20-минутном скользящем окне с размером журнала 1200 событий имеется единая скорость ввода 1200 в секунду. Последняя инструкция 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 секунды:

WITH SmootheningStep AS
(
SELECT
System.Timestamp() as time,
AVG(CAST(temperature as float)) as temp
FROM input
GROUP BY TUMBLINGWINDOW(second, 1)
),
AnomalyDetectionStep AS
(
SELECT
time,
temp,
AnomalyDetection_ChangePoint(temp, 80, 1200) 
OVER(LIMIT DURATION(minute, 20)) as ChangePointScores
FROM SmootheningStep
)

SELECT
time,
temp,
CAST(GetRecordPropertyValue(ChangePointScores, 'Score') as float) as
ChangePointScore,
CAST(GetRecordPropertyValue(ChangePointScores, 'IsAnomaly') as bigint) as
IsChangePointAnomaly

INTO output
FROM AnomalyDetectionStep

Пример с секционированных запросов для обучения отдельной модели на датчик:

WITH AnomalyDetectionStep AS
(
SELECT
sensorid,
System.Timestamp() as time,
CAST(temperature as float) as temp,
AnomalyDetection_ChangePoint(CAST(temperature as float), 80, 1200) 
OVER(PARTITION BY sensorid
LIMIT DURATION(minute, 20)) as ChangePointScores
FROM input
)

SELECT
CAST (sensorid as nvarchar(max)) as sensoridstring,
time,
temp,
CAST(GetRecordPropertyValue(ChangePointScores, 'Score') as float) as
ChangePointScore,
CAST(GetRecordPropertyValue(ChangePointScores, 'IsAnomaly') as bigint) as
IsChangePointAnomaly

INTO output
FROM AnomalyDetectionStep