AnomalyDetection_SpikeAndDip (Azure Stream Analytics)

Wykrywa tymczasowe anomalie w zdarzeniu szeregów czasowych.

Podstawowy model uczenia maszynowego używa algorytmu szacowania gęstości jądra adaptacyjnego.

Składnia

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

Argumenty

Scalar_expression

Kolumna zdarzeń lub obliczone pole, na którym model wykonuje wykrywanie anomalii. Dozwolone wartości tego parametru obejmują typy danych FLOAT lub BIGINT, które zwracają pojedynczą (skalarną) wartość.

Wyrażenie wieloznaczne * jest niedozwolone. Ponadto scalar_expression nie może zawierać innych funkcji analitycznych ani funkcji zewnętrznych.

Zaufania

Liczba procentowa z zakresu od 1,00 do 100 (włącznie), która określa wrażliwość modelu uczenia maszynowego. Im niższa pewność siebie, tym większa liczba wykrytych anomalii i odwrotnie. Zacznij od dowolnej liczby z zakresu od 70 do 90 i dostosuj tę wartość w oparciu o wyniki obserwowane podczas opracowywania lub testowania.

historySize

Liczba zdarzeń w oknie przewijania, z którego model stale uczy się i używa go do oceniania następnego zdarzenia w celu anomalii. Zazwyczaj powinno to reprezentować okres normalnego zachowania, aby umożliwić modelowi flagę kolejnej anomalii. Zacznij od wykształconego odgadnięcia przy użyciu dzienników historycznych i dostosuj się na podstawie wyników obserwowanych podczas opracowywania lub testowania.

tryb

Parametr ciągu, którego wartość to "skoki", "dips" lub "spikesanddips", aby wykryć tylko skoki, tylko spadki lub zarówno skoki, jak i spadki.

OVER ( [ partition_by_clause ] limit_duration_clause [when_clause])

partition_by_clause

Służy do partycjonowania trenowania modelu na podstawie określonej kolumny w zdarzeniach. Model stosuje te same ustawienia parametrów funkcji we wszystkich partycjach.

limit_duration_clause DURATION(jednostka, długość)

Rozmiar okna przewijania w usłudze Stream Analytics pod względem czasu. Zalecany rozmiar tego przedziału czasu jest odpowiednikiem czasu generowania historiiLiczba zdarzeń w stanie stałym.

when_clause

Określa warunek logiczny dla zdarzeń, które mają zostać zaakceptowane przez model w celu przeprowadzenia wykrywania anomalii. When_clause jest opcjonalna.

Typy zwracane

Funkcja zwraca zagnieżdżony rekord składający się z następujących kolumn:

IsAnomaly

BIGINT (0 lub 1) wskazujący, czy zdarzenie było nietypowe, czy nie.

Ocena

Obliczony wynik p-value (float) wskazujący, jak nietypowe jest zdarzenie. Niższe wyniki oznaczają mniejsze prawdopodobieństwo, że zdarzenie jest częścią tego samego rozkładu, a tym bardziej nietypowe jest.

Przykłady

W poniższym przykładzie przyjęto założenie, że jednolita szybkość wprowadzania wynosząca 1 zdarzenie na sekundę w 2-minutowym oknie przesuwnym z historią o rozmiarze 120 zdarzeń. Końcowa instrukcja SELECT wyodrębnia i generuje wynik oraz stan anomalii z poziomem ufności 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

Przykładowy strumień wejściowy, który jest jednolity przy użyciu okna wirowania 1 sekundy:

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

Przykład z partycjonowaną kwerendą w celu wytrenowania oddzielnego modelu na czujnik:

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