Rilevamento delle anomalie in Azure Stream Analytics

Disponibile sia nel cloud che in Azure IoT Edge, Analisi di flusso di Azure offre funzionalità predefinite di rilevamento anomalie basate su Machine Learning che possono essere usate per monitorare le due anomalie più comuni: temporanee e persistenti. Con le funzioni AnomalyDetection_SpikeAndDip e AnomalyDetection_ChangePoint , è possibile eseguire il rilevamento anomalie direttamente nel processo di Analisi di flusso.

I modelli di Machine Learning presuppongono una serie temporale campionata in modo uniforme. Se la serie temporale non è uniforme, è possibile inserire un passaggio di aggregazione con una finestra a scorrimento prima di chiamare il rilevamento di anomalie.

Le operazioni di Machine Learning non supportano attualmente tendenze di stagionalità o correlazioni multi-variate.

Il video seguente illustra come rilevare un'anomalia in tempo reale usando le funzioni di Machine Learning in Analisi di flusso di Azure.

In genere, l'accuratezza del modello migliora con più dati nella finestra scorrevole. I dati nella finestra temporale scorrevole specificata vengono considerati come parte dell'intervallo normale di valori per tale intervallo di tempo. Il modello considera solo la cronologia degli eventi nella finestra temporale scorrevole per verificare se l'evento corrente è anomalo. Man mano che la finestra temporale scorrevole scorre, i vecchi valori vengono rimossi dal training del modello.

Le funzioni operano stabilendo una certa normalità in base a ciò che hanno visto finora. I valori anomali vengono identificati confrontandoli con la norma stabilita, all'interno del livello di fiducia. Le dimensioni della finestra devono essere basate sugli eventi minimi necessari per eseguire il training del modello per il comportamento normale in modo che, quando si verifica un'anomalia, sia in grado di riconoscerlo.

Il tempo di risposta del modello aumenta con le dimensioni della cronologia perché deve essere confrontato con un numero maggiore di eventi precedenti. È consigliabile includere solo il numero necessario di eventi per ottenere prestazioni migliori.

Le lacune nella serie temporale possono essere il risultato del modello che non riceve eventi in determinati momenti. Questa situazione viene gestita da Analisi di flusso usando la logica di imputazione. Le dimensioni della cronologia, nonché una durata temporale, per la stessa finestra temporale scorrevole vengono usate per calcolare la frequenza media di arrivo degli eventi.

Un generatore di anomalie disponibile qui può essere usato per inserire un hub IoT con dati con modelli di anomalie diversi. È possibile configurare un processo di Analisi di flusso di Azure con queste funzioni di rilevamento anomalie per leggere da questo hub IoT e rilevare le anomalie.

Le anomalie temporanee in un flusso di eventi delle serie temporali sono note come picchi e dips. È possibile monitorare picchi e flessioni usando l'operatore basato su Machine Learning , AnomalyDetection_SpikeAndDip.

Nella stessa finestra temporale scorrevole, se un secondo picco è inferiore al primo, il punteggio calcolato per il picco più piccolo probabilmente non è sufficientemente significativo rispetto al punteggio per il primo picco all'interno del livello di attendibilità specificato. È possibile provare a ridurre il livello di attendibilità del modello per rilevare tali anomalie. Tuttavia, se si inizia a ricevere troppi avvisi, è possibile usare un intervallo di confidenza più elevato.

La query di esempio seguente presuppone una frequenza di input uniforme di un evento al secondo in una finestra temporale scorrevole di 2 minuti con una cronologia di 120 eventi. L'istruzione SELECT finale estrae e restituisce il punteggio e lo stato dell'anomalia con un livello di attendibilità del 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

Le anomalie persistenti in un flusso di eventi di una serie temporale sono modifiche nella distribuzione dei valori nel flusso di eventi, ad esempio modifiche al livello e tendenze. In Analisi di flusso tali anomalie vengono rilevate usando l'operatore di AnomalyDetection_ChangePoint basato su Machine Learning.

Le modifiche persistenti durano molto più a lungo dei picchi e delle immersioni e potrebbero indicare eventi irreversibili. Le modifiche persistenti non sono in genere visibili all'occhio nudo, ma possono essere rilevate con l'operatore AnomalyDetection_ChangePoint .

L'immagine seguente è un esempio di modifica del livello:

L'immagine seguente è un esempio di modifica della tendenza:

La query di esempio seguente presuppone una frequenza di input uniforme di un evento al secondo in una finestra temporale scorrevole di 20 minuti con dimensioni della cronologia di 1.200 eventi. L'istruzione SELECT finale estrae e restituisce il punteggio e lo stato delle anomalie con un livello di attendibilità pari a 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

Le prestazioni di questi modelli dipendono dalle dimensioni della cronologia, dalla durata della finestra, dal caricamento degli eventi e dall'uso del partizionamento a livello di funzione. In questa sezione vengono descritte queste configurazioni e forniti esempi su come mantenere tassi di inserimento di 1 K, 5 K e 10 K eventi per secondo.

• Dimensioni cronologia : questi modelli vengono eseguiti in modo lineare con le dimensioni della cronologia. Più lunga è la dimensione della cronologia, più i modelli impiegano per assegnare un punteggio a un nuovo evento. Il motivo è che i modelli confrontano il nuovo evento con ognuno degli eventi precedenti nel buffer della cronologia.
• Durata finestra : la durata della finestra deve riflettere il tempo necessario per ricevere tutti gli eventi specificati dalle dimensioni della cronologia. Senza questo numero di eventi nella finestra, Analisi di flusso di Azure imputa valori mancanti. Di conseguenza, il consumo della CPU è una funzione della dimensione della cronologia.
• Carico eventi : maggiore è il carico degli eventi, maggiore è il lavoro eseguito dai modelli, che influisce sull'utilizzo della CPU. Il processo può essere ridimensionato rendendolo perfettamente parallelo, presupponendo che sia opportuno usare più partizioni di input per la logica di business.
• Partizionamento a livello di funzione - Il partizionamento a livello di funzione viene eseguito utilizzando PARTITION BY all'interno della chiamata di funzione di rilevamento delle anomalie. Questo tipo di partizionamento comporta un sovraccarico, perché lo stato deve essere mantenuto per più modelli contemporaneamente. Il partizionamento a livello di funzione viene usato in scenari come il partizionamento a livello di dispositivo.

Le dimensioni della cronologia, la durata della finestra e il carico totale degli eventi sono correlati nel modo seguente:

windowDuration (in ms) = 1000 * historySize / (numero totale di eventi di input al secondo / Numero di partizioni di input)

Quando si partiziona la funzione in base a deviceId, aggiungere "PARTITION BY deviceId" alla chiamata di funzione di rilevamento anomalie.

La tabella seguente include le osservazioni relative alla capacità di trasmissione per un singolo nodo (sei unità di servizio) nel caso non partizionato:

La tabella seguente include le osservazioni sul throughput per un singolo nodo (sei SU) per il caso partizionato.

Il codice di esempio per eseguire le configurazioni non partizionate precedenti si trova nel repository Streaming At Scale di Esempi di Azure. Il codice crea un processo di analisi di flusso senza partizionamento a livello di funzione, che usa Hub eventi come input e output. Il carico di input viene generato usando i client di test. Ogni evento di input è un documento JSON di 1 KB. Gli eventi simulano un dispositivo IoT che invia dati JSON (per un massimo di 1 K dispositivi). Le dimensioni della cronologia, la durata della finestra e il carico totale degli eventi sono variati in due partizioni di input.

Usare il riquadro Metriche nel processo di Analisi di flusso di Azure per identificare i colli di bottiglia nella pipeline. Vedere Eventi di input/output per la velocità effettiva e "Ritardo limite" o Eventi con backlog per verificare se il processo è in grado di mantenere la frequenza di input. Per le metriche dell'hub eventi, cercare Richieste limitate e modificare di conseguenza le unità di soglia. Per le metriche di Azure Cosmos DB, vedere Numero massimo di unità richiesta al secondo usate per intervallo di chiavi di partizione in "Velocità effettiva" per assicurarsi che gli intervalli delle chiavi di partizione siano usati in modo uniforme. Per il database SQL di Azure, monitorare I/O LOG e CPU.

• Introduzione ad Analisi dei flussi di Azure
• Introduzione all'uso di Analisi dei flussi di Azure
• Ridimensionare i processi di Analisi dei flussi di Azure
• Informazioni di riferimento sul linguaggio di query di Analisi di flusso di Azure
• Informazioni di riferimento sulle API REST di gestione di Analisi di flusso di Azure