AnomalyDetection_SpikeAndDip (Azure Stream Analytics)

Artigo
07/12/2023

Deteta anomalias temporárias num evento de série temporal.

O modelo de machine learning subjacente utiliza o algoritmo de estimativa de densidade de kernel adaptável.

Sintaxe

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

Argumentos

scalar_expression

A coluna de eventos ou o campo calculado sobre o qual o modelo executa a deteção de anomalias. Os valores permitidos para este parâmetro incluem tipos de dados FLOAT ou BIGINT que devolvem um único valor (escalar).

A expressão de caráter universal * 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 entre 1,00 e 100 (inclusive) que define a sensibilidade do modelo de machine learning. Quanto menor for a confiança, maior é o número de anomalias detetadas e vice-versa. Comece a partir de 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 numa janela deslizante que o modelo aprende continuamente e utiliza para classificar o evento seguinte para anómalo. Normalmente, isto deve representar o período de tempo do comportamento normal para permitir que o modelo sinalize uma anomalia subsequente. Comece com um palpite educado através de registos históricos e ajuste-se com base nos resultados observados no desenvolvimento ou teste.

modo

Um parâmetro de cadeia cujo valor é "picos", "dips" ou "spikesanddips", para detetar apenas picos, apenas quedas ou picos e quedas, respetivamente.

OVER ( [ partition_by_clause ] limit_duration_clause [when_clause])

partition_by_clause

Utilizado para particionar a preparação de um modelo com base numa coluna específica nos eventos. O modelo aplica as mesmas definições de parâmetro de função em todas as partições.

limit_duration_clause DURAÇÃO(unidade; comprimento)

O tamanho da janela deslizante no Stream Analytics em termos de tempo. O tamanho recomendado desta janela de tempo é o equivalente ao tempo que demora a gerar o históricoDimensionar o número de eventos num estado estável.

when_clause

Especifica a condição booleana para que os eventos sejam aceites pelo modelo para executar a deteção de anomalias. O when_clause é opcional.

Tipos de Retorno

A função devolve um registo aninhado composto pelas seguintes colunas:

IsAnomaly

Um BIGINT (0 ou 1) que indica se o evento foi anómalo ou não.

Pontuação

A classificação de p-value calculada (float) que indica o quão anómalo é um evento. As classificações mais baixas significam uma menor probabilidade de o evento fazer parte da mesma distribuição e, portanto, quanto mais anómalo for.

Exemplos

O exemplo seguinte pressupõe uma taxa de entrada uniforme de 1 evento por segundo numa janela deslizante de 2 minutos com um tamanho histórico de 120 eventos. A instrução SELECT final extrai e produz a classificação e o estado da anomalia com um nível de confiança de 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

Exemplo com um fluxo de entrada não uniforme que é uniforme com 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_SpikeAndDip(temp, 95, 120, 'spikesanddips') 
        OVER(LIMIT DURATION(second, 120)) as SpikeAndDipScores
    FROM SmootheningStep
)
SELECT
    time,
    temp,
    CAST(GetRecordPropertyValue(SpikeAndDipScores, 'Score') AS FLOAT) As
    SpikeAndDipScore,
    CAST(GetRecordPropertyValue(SpikeAndDipScores, 'IsAnomaly') AS BIGINT) AS
    IsSpikeAndDipAnomaly
INTO output
FROM AnomalyDetectionStep

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

WITH AnomalyDetectionStep AS
(
    SELECT
        sensorid,
        System.Timestamp() AS time,
        CAST(temperature AS FLOAT) AS temp,
        AnomalyDetection_SpikeAndDip(CAST(temperature AS FLOAT), 95, 120, 'spikesanddips')
            OVER(PARTITION BY sensorid LIMIT DURATION(second, 120)) AS SpikeAndDipScores
    FROM input
)
SELECT
    CAST (sensorid AS NVARCHAR(max)) AS sensoridstring,
    time,
    temp,
    CAST(GetRecordPropertyValue(SpikeAndDipScores, 'Score') as float) AS
    SpikeAndDipScore,
    CAST(GetRecordPropertyValue(SpikeAndDipScores, 'IsAnomaly') AS BIGINT) AS
    IsSpikeAndDipAnomaly
INTO output
FROM AnomalyDetectionStep

Share via