AnomalyDetection_ChangePoint

✅ Azure Stream Analytics ✅ Fabric Eventstream

Erkennt persistente Anomalien in einem Zeitreihenereignisdatenstrom. Das zugrunde liegende Machine Learning-Modell verwendet den Exchangeability Martingales-Algorithmus.

Syntax

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

Argumente

scalar_expression

Die Ereignisspalte oder das berechnete Feld, über das das Modell Anomalieerkennung durchführt. Zulässige Werte für diesen Parameter umfassen FLOAT- oder BIGINT-Datentypen, die einen einzelnen (skalaren) Wert zurückgeben.

Der Wildcardausdruck * ist nicht zulässig. Außerdem können scalar_expression keine anderen Analysefunktionen oder externen Funktionen enthalten.

Vertrauen

Eine Prozentzahl von 1,00 bis 100 (einschließlich), die die Vertraulichkeit des Machine Learning-Modells festlegt. Je niedriger das Vertrauen, desto höher ist die Anzahl der erkannten Anomalien und umgekehrt. Beginnen Sie mit einer beliebigen Zahl zwischen 70 und 90, und passen Sie diese basierend auf den ergebnissen an, die bei der Entwicklung oder beim Testen beobachtet wurden.

historySize

Die Anzahl der Ereignisse in einem Gleitfenster, von denen das Modell kontinuierlich lernt und verwendet, um das nächste Ereignis für anomalienhaft zu bewerten. In der Regel sollte dies den Zeitraum des normalen Verhaltens darstellen, damit das Modell eine nachfolgende Anomalie kennzeichnen kann. Beginnen Sie mit einer gebildeten Vermutung mithilfe von historischen Protokollen, und passen Sie basierend auf den ergebnissen, die bei der Entwicklung oder beim Test beobachtet wurden.

OVER ([ partition_by_clause ] limit_duration_clause [when_clause])

partition_by_clause

Wird verwendet, um die Schulung eines Modells basierend auf einer bestimmten Spalte in den Ereignissen zu partitionieren. Das Modell wendet die gleichen Funktionsparametereinstellungen für alle Partitionen an.

limit_duration_clause DURATION(Einheit; Länge)

Die Größe des gleitenden Fensters in Stream Analytics im Hinblick auf die Zeit. Die empfohlene Größe dieses Zeitfensters entspricht der Zeit, die zum Generieren der Anzahl von Ereignissen im stabilen Zustand benötigt wird.

when_clause

Gibt eine boolesche Bedingung für die Ereignisse an, die für das Modell bereitgestellt werden sollen, um Anomalieerkennung durchzuführen. Die when_clause ist optional.

Rückgabetypen

Die Funktion gibt einen geschachtelten Datensatz zurück, der aus den folgenden Spalten besteht:

IsAnomaly

Ein BIGINT (0 oder 1), der angibt, ob das Ereignis anomale oder nicht.

Ergebnis

Die berechnete Martingale-Bewertung (float), die angibt, wie anomalien ein Ereignis ist. Diese Bewertung wächst exponentiell mit anomalen Werten.

Beispiele

Im folgenden Abfragebeispiel geht die erste Abfrage alle 5 Minuten von einem Ereignis aus, und die zweite Abfrage setzt jedes Sekunde ein Ereignis voraus. Das Konfidenzniveau wird für beide Modelle auf 75 festgelegt.

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

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

Beispiel für eine einheitliche Eingaberate von 1 Ereignis pro Sekunde in einem 20-minütigen Gleitfenster mit einer Verlaufsgröße von 1200 Ereignissen. Die abschließende SELECT-Anweisung extrahiert und gibt die Punktzahl und den Anomaliestatus mit einem Konfidenzniveau von 80%aus.

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

Beispiel mit einem nicht einheitlichen Eingabestrom, der mithilfe eines Sturzfensters von 1 Sekunde einheitlich gestaltet wird:

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

Beispiel mit einer partitionierten Abfrage zum Trainieren eines separaten Modells pro 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