DetecçãoDeAnomalias_PontoDeMudança

✅ Fluxo de eventos do Azure Stream Analytics ✅ Fabric

Detecta anomalias persistentes em um fluxo de eventos de série temporal. O modelo de machine learning subjacente usa o algoritmo Exchangeability Martingales.

Sintaxe

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

Argumentos

scalar_expression

A coluna de evento ou o campo computado sobre o qual o modelo executa a detecção de anomalias. Os valores permitidos para esse parâmetro incluem tipos de dados FLOAT ou BIGINT que retornam um único valor (escalar).

A expressão curinga * não é permitida. Além disso, scalar_expression não pode conter outras funções analíticas ou funções externas.

confiança

Um número percentual de 1,00 a 100 (inclusive) que define a confidencialidade do modelo de machine learning. Quanto menor a confiança, maior o número de anomalias detectadas e vice-versa. Comece com um número arbitrário entre 70 e 90 e ajuste-o com base nos resultados observados no desenvolvimento ou teste.

historySize

O número de eventos em uma janela deslizante com a qual o modelo aprende continuamente e usa para pontuar o próximo evento para anomalias. Normalmente, isso deve representar o período de tempo de comportamento normal para permitir que o modelo sinalize uma anomalia subsequente. Comece com uma adivinhação educada usando logs históricos e ajuste com base nos resultados observados no desenvolvimento ou teste.

OVER ([ partition_by_clause] limit_duration_clause [when_clause])

partition_by_clause

Usado para particionar o treinamento de um modelo com base em uma coluna específica nos eventos. O modelo aplica as mesmas configurações de parâmetro de função em todas as partições.

limit_duration_clause DURATION(unit, length)

O tamanho da janela deslizante no Stream Analytics em termos de tempo. O tamanho recomendado dessa janela de tempo é o equivalente ao tempo necessário para gerar o número de eventos historySize em estado estável.

when_clause

Especifica a condição booliana para os eventos a serem fornecidos ao modelo para executar a detecção de anomalias. O when_clause é opcional.

Tipos de retorno

A função retorna um registro aninhado composto das seguintes colunas:

IsAnomaly

UM BIGINT (0 ou 1) que indica se o evento era anômíncrono ou não.

Pontuação

A pontuação computada de Martingale (float) que indica o quão anômalo é um evento. Essa pontuação aumenta exponencialmente com valores anômalos.

Exemplos

No exemplo de consulta a seguir, a primeira consulta pressupõe um evento a cada 5 minutos e a segunda consulta pressupõe um evento a cada segundo. O nível de confiança é definido em 75 para ambos os modelos.

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

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

Exemplo supondo uma taxa de entrada uniforme de 1 evento por segundo em uma janela deslizante de 20 minutos com um tamanho histórico de 1200 eventos. A instrução final SELECT extrai e gera o score e o status da anomalia com um nível de confiança de 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

Exemplo com um fluxo de entrada não uniforme que se torna uniforme usando uma janela em cascata de 1 segundo:

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

Exemplo com uma consulta particionada para treinar um modelo separado por sensor:

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