Detecção de anomalias no Azure Stream AnalyticsAnomaly detection in Azure Stream Analytics

Disponível na nuvem e no Azure IoT Edge, o Azure Stream Analytics oferece recursos internos de detecção de anomalias baseados em aprendizado de máquina que podem ser usados para monitorar as duas anomalias que ocorrem com mais frequência: temporárias e persistentes.Available in both the cloud and Azure IoT Edge, Azure Stream Analytics offers built-in machine learning based anomaly detection capabilities that can be used to monitor the two most commonly occurring anomalies: temporary and persistent. Com as funções AnomalyDetection_SpikeAndDip e AnomalyDetection_ChangePoint, você pode detectar anomalias diretamente no seu trabalho do Stream Analytics.With the AnomalyDetection_SpikeAndDip and AnomalyDetection_ChangePoint functions, you can perform anomaly detection directly in your Stream Analytics job.

Os modelos de machine learning supõem uma série temporal incluída na amostra de maneira uniforme.The machine learning models assume a uniformly sampled time series. Se a série temporal não for uniforme, insira uma etapa de agregação com uma janela em cascata antes de chamar a detecção de anomalias.If the time series is not uniform, you may insert an aggregation step with a tumbling window prior to calling anomaly detection.

As operações de aprendizado de máquina não suportam as tendências de sazonalidade ou você com vários correlações neste momento.The machine learning operations do not support seasonality trends or multi-variate correlations at this time.

Comportamento do modeloModel behavior

De modo geral, a precisão do modelo melhora com mais dados na janela deslizante.Generally, the model's accuracy improves with more data in the sliding window. Os dados na janela deslizante especificada são tratados como parte de seu intervalo de valores normal para o período.The data in the specified sliding window is treated as part of its normal range of values for that time frame. O modelo considera o histórico de eventos ao longo da janela deslizante apenas para verificar se o evento atual é anormal.The model only considers event history over the sliding window to check if the current event is anomalous. Conforme a janela deslizante se move, os valores antigos são removidos do treinamento do modelo.As the sliding window moves, old values are evicted from the model’s training.

As funções operam estabelecendo um determinado valor normal com base no que foi observado até então.The functions operate by establishing a certain normal based on what they have seen so far. As exceções são identificadas pela comparação em relação ao normal estabelecido, no nível de confiança.Outliers are identified by comparing against the established normal, within the confidence level. O tamanho da janela deve se basear nos eventos mínimos necessários para treinar o modelo para comportamento normal, de modo que ele esteja apto a reconhecê-la quando uma anomalia ocorrer.The window size should be based on the minimum events required to train the model for normal behavior so that when an anomaly occurs, it would be able to recognize it.

Tempo de resposta do modelo aumenta com o tamanho do histórico, porque ele precisa ser comparada com um número maior de eventos passados.The model's response time increases with history size because it needs to compare against a higher number of past events. É recomendável incluir apenas o número necessário de eventos para obter melhor desempenho.It is recommended to only include the necessary number of events for better performance.

As lacunas na série temporal podem ser um resultado do não recebimento de eventos pelo modelo em determinados pontos no tempo.Gaps in the time series can be a result of the model not receiving events at certain points in time. Essa situação é tratada pela análise de Stream usando imputação lógica.This situation is handled by Stream Analytics using imputation logic. O tamanho do histórico e a duração para a mesma janela deslizante são usados para calcular a taxa média na qual os eventos são esperados.The history size, as well as a time duration, for the same sliding window is used to calculate the average rate at which events are expected to arrive.

Um gerador de anomalias disponível aqui pode ser usado para alimentar um Iot Hub com os dados com padrões diferentes de anomalias.An anomaly generator available here can be used to feed an Iot Hub with data with different anomaly patterns. Um trabalho ASA pode ser configurado com essas funções de detecção de anomalias para ler esse Iot Hub e detectar anomalias.An ASA job can be set up with these anomaly detection functions to read from this Iot Hub and detect anomalies.

Pico e quedaSpike and dip

As anomalias temporárias em um fluxo de eventos de série temporal são conhecidas como picos e quedas.Temporary anomalies in a time series event stream are known as spikes and dips. Os picos e quedas podem ser monitorados usando o operador baseado em Machine Learning, AnomalyDetection_SpikeAndDip.Spikes and dips can be monitored using the Machine Learning based operator, AnomalyDetection_SpikeAndDip.

Exemplo de anomalia de pico e queda

Na mesma janela deslizante, se um segundo pico for menor que o primeiro, a pontuação calculada para o menor pico provavelmente não será significativa o suficiente em comparação com a pontuação para o primeiro pico no nível de confiança especificado.In the same sliding window, if a second spike is smaller than the first one, the computed score for the smaller spike is probably not significant enough compared to the score for the first spike within the confidence level specified. Você pode tentar diminuir o nível de confiança do modelo para detectar essas anomalias.You can try decreasing the model's confidence level to detect such anomalies. No entanto, se começar a receber muitos alertas, é possível usar um intervalo de confiança maior.However, if you start to get too many alerts, you can use a higher confidence interval.

O exemplo de consulta a seguir pressupõe uma taxa uniforme de entrada de um evento por segundo em uma janela deslizante de 2 minutos com um histórico de 120 eventos.The following example query assumes a uniform input rate of one event per second in a 2-minute sliding window with a history of 120 events. A instrução SELECT final extrai e gera a pontuação e o status da anomalia com um nível de confiança de 95%.The final SELECT statement extracts and outputs the score and anomaly status with a confidence level of 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

Ponto de alteraçãoChange point

As anomalias persistentes em um fluxo de eventos de série temporal são alterações na distribuição de valores no fluxo de eventos, como alterações e tendências no nível.Persistent anomalies in a time series event stream are changes in the distribution of values in the event stream, like level changes and trends. No Stream Analytics, tais anomalias são detectadas usando o operador AnomalyDetection_ChangePoint baseado em Machine Learning.In Stream Analytics, such anomalies are detected using the Machine Learning based AnomalyDetection_ChangePoint operator.

As alterações persistentes duram muito mais que picos e quedas e podem indicar eventos catastróficos.Persistent changes last much longer than spikes and dips and could indicate catastrophic event(s). As alterações persistentes geralmente não são vistas a olho nu, mas podem ser detectadas com o operador AnomalyDetection_ChangePoint.Persistent changes are not usually visible to the naked eye, but can be detected with the AnomalyDetection_ChangePoint operator.

A seguinte imagem é um exemplo de alteração no nível:The following image is an example of a level change:

Exemplo de anomalia de alteração no nível

A seguinte imagem é um exemplo de alteração na tendência:The following image is an example of a trend change:

Exemplo de anomalia de alteração na tendência

O exemplo de consulta a seguir pressupõe uma taxa uniforme de entrada de um evento por segundo em uma janela deslizante de 20 minutos com um tamanho de histórico de 1200 eventos.The following example query assumes a uniform input rate of one event per second in a 20-minute sliding window with a history size of 1200 events. A instrução SELECT final extrai e gera a pontuação e o status da anomalia com um nível de confiança de 80%.The final SELECT statement extracts and outputs the score and anomaly status with a confidence level of 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

Características de desempenhoPerformance characteristics

O desempenho desses modelos depende do tamanho do histórico, duração da janela, a carga de eventos, e se o particionamento de nível de função é usado.The performance of these models depends on the history size, window duration, event load, and whether function level partitioning is used. Esta seção aborda essas configurações e fornece exemplos de como manter as taxas de ingestão de 1K, K 5 e 10 mil eventos por segundo.This section discusses these configurations and provides samples for how to sustain ingestion rates of 1K, 5K and 10K events per second.

  • Tamanho do histórico -esses modelos executam linearmente com tamanho do histórico.History size - These models perform linearly with history size. Quanto maior for o tamanho do histórico, mais tempo os modelos de levam para um novo evento de pontuação.The longer the history size, the longer the models take to score a new event. Isso ocorre porque os modelos de comparam o novo evento com cada um dos últimos eventos no buffer de histórico.This is because the models compare the new event with each of the past events in the history buffer.
  • Duração da janela - o duração da janela deve refletir quanto tempo leva para receber quantos eventos conforme especificado pelo tamanho do histórico.Window duration - The Window duration should reflect how long it takes to receive as many events as specified by the history size. Sem que vários eventos da janela, o Azure Stream Analytics faria imputar valores ausentes.Without that many events in the window, Azure Stream Analytics would impute missing values. Portanto, o consumo de CPU é uma função do tamanho do histórico.Hence, CPU consumption is a function of the history size.
  • Carga de eventos - quanto maior a carga do evento, mais trabalho que é executada pelos modelos, que afeta o consumo de CPU.Event load - The greater the event load, the more work that is performed by the models, which impacts CPU consumption. O trabalho pode ser escalado horizontalmente, tornando embaraçosamente paralelos, supondo que ele faça sentido para lógica de negócios para usar mais partições de entrada.The job can be scaled out by making it embarrassingly parallel, assuming it makes sense for business logic to use more input partitions.
  • Particionamento de nível de função - particionamento de nível de função é feito usando PARTITION BY dentro da chamada de função de detecção de anomalias.Function level partitioning - Function level partitioning is done by using PARTITION BY within the anomaly detection function call. Esse tipo de particionamento adiciona uma sobrecarga, como o estado precisa ser mantido para vários modelos ao mesmo tempo.This type of partitioning adds an overhead, as state needs to be maintained for multiple models at the same time. Particionamento de nível de função é usado em cenários como o particionamento de nível de dispositivo.Function level partitioning is used in scenarios like device level partitioning.

RelaçãoRelationship

O tamanho do histórico, duração da janela e carga total de eventos são relacionadas da seguinte maneira:The history size, window duration, and total event load are related in the following way:

windowDuration (em ms) = 1000 * historySize / (Total eventos de entrada por segundo / contagem de partições de entrada)windowDuration (in ms) = 1000 * historySize / (Total Input Events Per Sec / Input Partition Count)

Quando a função de particionamento por deviceId, adicione "PARTITION BY deviceId" para a chamada de função de detecção de anomalias.When partitioning the function by deviceId, add “PARTITION BY deviceId” to the anomaly detection function call.

ObservaçõesObservations

A tabela a seguir inclui as observações de taxa de transferência para um único nó (6 UAS) para o caso não particionado:The following table includes the throughput observations for a single node (6 SU) for the non-partitioned case:

Tamanho do histórico (eventos)History size (events) Duração da janela (ms)Window duration (ms) Eventos de entrada total por segundoTotal input events per sec
6060 5555 2,2002,200
600600 728728 1,6501,650
6.0006,000 10,91010,910 1,1001,100

A tabela a seguir inclui as observações de taxa de transferência para um único nó (6 UAS) para o caso particionado:The following table includes the throughput observations for a single node (6 SU) for the partitioned case:

Tamanho do histórico (eventos)History size (events) Duração da janela (ms)Window duration (ms) Eventos de entrada total por segundoTotal input events per sec Contagem de dispositivosDevice count
6060 1,0911,091 1,1001,100 1010
600600 10,91010,910 1,1001,100 1010
6.0006,000 218,182218,182 <550<550 1010
6060 21,81921,819 550550 100100
600600 218,182218,182 550550 100100
6.0006,000 2,181,8192,181,819 <550<550 100100

Código de exemplo para executar as configurações acima não particionada está localizado na repositório de Streaming em escala de amostras do Azure.Sample code to run the non-partitioned configurations above is located in the Streaming At Scale repo of Azure Samples. O código cria um trabalho do stream analytics com nenhuma função nível particionamento, que usa o Hub de eventos como entrada e saída.The code creates a stream analytics job with no function level partitioning, which uses Event Hub as input and output. A carga de entrada é gerada usando clientes de teste.The input load is generated using test clients. Cada evento de entrada é um documento de json de 1KB.Each input event is a 1KB json document. Eventos de simulam um dispositivo de IoT enviando dados JSON (para dispositivos de até 1 K).Events simulate an IoT device sending JSON data (for up to 1K devices). O tamanho do histórico, duração da janela e total de eventos de carga variam ao longo de 2 partições de entrada.The history size, window duration, and total event load are varied over 2 input partitions.

Observação

Para obter uma estimativa mais precisa, personalize os exemplos para ajustar seu cenário.For a more accurate estimate, customize the samples to fit your scenario.

Identificando afunilamentosIdentifying bottlenecks

Use o painel de métricas no trabalho do Azure Stream Analytics para identificar gargalos no seu pipeline.Use the Metrics pane in your Azure Stream Analytics job to identify bottlenecks in your pipeline. Revisão eventos de entrada/saída taxa de transferência e "Atraso de marca d'água" ou eventos com lista de pendências para ver se o trabalho está acompanhando a taxa de entrada.Review Input/Output Events for throughput and "Watermark Delay" or Backlogged Events to see if the job is keeping up with the input rate. Para métricas do Hub de eventos, procure solicitações limitadas e ajustar as unidades de limite adequadamente.For Event Hub metrics, look for Throttled Requests and adjust the Threshold Units accordingly. Para métricas do Cosmos DB, examine máximo de RU/s consumidas por intervalo de chaves de partição na taxa de transferência para garantir que sua partição de intervalos de chaves são consumidos uniformemente.For Cosmos DB metrics, review Max consumed RU/s per partition key range under Throughput to ensure your partition key ranges are uniformly consumed. Para o Azure SQL DB, monitore e/s de Log e CPU.For Azure SQL DB, monitor Log IO and CPU.

Detecção de anomalias usando o machine learning no Azure Stream AnalyticsAnomaly detection using machine learning in Azure Stream Analytics

O vídeo a seguir demonstra como detectar uma anomalia em tempo real usando funções de machine learning no Azure Stream Analytics.The following video demonstrates how to detect an anomaly in real time using machine learning functions in Azure Stream Analytics.

Próximas etapasNext steps