Negli ultimi anni si sta consolidando sempre più la necessità di raccogliere dati ed effettuare analisi su di essi. Azure Stream Analytics rappresenta il primo step di un'architettura che abbia il compito di aggregare e trasformare dati provenienti da dispositivi o altri database, per produrre aggregazioni da scrivere poi su altri output, come database o code.

Oltre ad aggregare e trasformare dati però, è utile poter applicare modelli di reti neurali che ci permettano di ottenere più valore da questi dati. Un esempio è quello di capire se si verificano anomalie e poter reagire nei più svariati dei modi. Implementare questi algoritmi non è affatto facile, perché sono richieste conoscenze di data scientist e la costruzione di motori che tengano in vita questi modelli.

Di recente è stato introdotta la possibilità di implementare questa funzionalità usando un modello già pronto all'uso direttamente in Stream Analytics e con poche righe di SQL statement ottenere un indice che valuti i dati da noi passati.

Questo è possibile grazie all'introduzione di due funzioni:
- AnomalyDetection_ChangePoint: identifica un cambio di un trend nell'intervallo stabilito;
- AnomalyDetection_SpikeAndDip: identifica un picco o una flessione improvvisa nell'intervallo stabilito.
Queste due funzioni ci permettono di passare un valore, scalare oppure composto secondo il significato che vogliamo dare, e di ottenere un valore binario per indicare se è stato rilevato un cambio o un'anomalia. Il modello si adatta automaticamente nel tempo sulla base dei dati passati e restituisce uno score di affidabilità sull'anomalia rilevata.

Ipotizziamo quindi avere in input valori rilevati da un termostato. Su di esso possiamo costruire una CTE che oltre alla temperatura e al device, chiami la funzione per restituire un oggetto composto che contenga la valutazione di eventuali picchi o flessioni sulla temperature stessa.

WITH AnomalyDetectionStep AS
(
    SELECT
        date,
        temperature,
        deviceId,
        AnomalyDetection_SpikeAndDip(temperature, 95, 360, 'spikesanddips')
        OVER(PARTITION BY deviceId LIMIT DURATION(hour, 1)) AS SpikeAndDipScores
    FROM input Partition By deviceId
)

La funzione AnomalyDetection_SpikeAndDip vuole il valore da valutare, un livello di confidenza da 1 a 100 da adottare (più è basso più anomalie vengono trovate), il numero di eventi/valori da considerare nel passato oltre a quello corrente e quale modalità adottare (spikesanddips, dips o spikes).

A questo punto nello statement di produzione dell'output, possiamo usare la funzione GetRecordPropertyValue per estrarre il risultato della valutazione e lo score. Nell'esempio seguente estraiamo questi dati ed emettiamo solo i valori che risultano essere anomali.

SELECT
    date as RowKey, deviceId as PartitionKey, temperature,
    ROUND(CAST(GetRecordPropertyValue(SpikeAndDipScores, 'Score') as FLOAT), 2) AS score,
    CAST(GetRecordPropertyValue(SpikeAndDipScores, 'IsAnomaly') AS BIGINT) AS anomaly
INTO alarms
FROM AnomalyDetectionStep Partition By deviceId
WHERE CAST(GetRecordPropertyValue(SpikeAndDipScores, 'IsAnomaly') AS BIGINT) = 1
    AND CAST(GetRecordPropertyValue(SpikeAndDipScores, 'Score') as FLOAT) < 0.05

Oltre ad usare la funzione, negli snippet precedenti viene anche usata una partizione per device in modo da produrre set diversi di training sull'anomalie che cambiano a seconda del termostato che riporta i dati. Anche la partizione ha una finestra temporale da tenere in considerazione che in genere si allinea al numero di punti che la funzione necessita.