Exemplo: detecção de anomalias de dados em um fluxo (função RANDOM_CUT_FOREST)

O Amazon Kinesis Data Analytics fornece uma função (RANDOM_CUT_FOREST) que pode atribuir uma pontuação de anomalias a cada registro de acordo com os valores nas colunas numéricas. Para obter mais informações, consulte RANDOM_CUT_FORESTFunções em Amazon Managed Service for Apache Flink SQL Reference.

Neste exercício, você grava o código de aplicativo para atribuir uma pontuação de anomalias a registros na origem de streaming do aplicativo. Para configurar o aplicativo, faça o seguinte:

1. Configurar uma origem de streaming: configure um fluxo de dados do Kinesis e grave dados heartRate de amostra, como mostrado a seguir:

   {"heartRate": 60, "rateType":"NORMAL"}
   ...
   {"heartRate": 180, "rateType":"HIGH"}

   O procedimento fornece um script Python para preencher o stream. Os valores heartRate são gerados aleatoriamente, com 99% dos registros com valores heartRate entre 60 e 100, e apenas 1% de valores heartRate entre 150 e 200. Assim, os registros que têm valores heartRate entre 150 e 200 são anomalias.

2. Configurar entrada: usando o console, crie um aplicativo do Kinesis Data Analytics e configure a entrada de aplicativo mapeando a origem de streaming para um fluxo no aplicativo (SOURCE_SQL_STREAM_001). Quando o aplicativo é iniciado, o Kinesis Data Analytics lê continuamente a origem do streaming e insere registros no streaming do aplicativo de entrada.

3. Especificar o código de aplicativo – O exemplo usa o seguinte código de aplicativo:

   --Creates a temporary stream.
   CREATE OR REPLACE STREAM "TEMP_STREAM" (
   	        "heartRate"        INTEGER,
   	        "rateType"         varchar(20),
   	        "ANOMALY_SCORE"    DOUBLE);
   
   --Creates another stream for application output.	        
   CREATE OR REPLACE STREAM "DESTINATION_SQL_STREAM" (
   	        "heartRate"        INTEGER,
   	        "rateType"         varchar(20),
   	        "ANOMALY_SCORE"    DOUBLE);
   
   -- Compute an anomaly score for each record in the input stream
   -- using Random Cut Forest
   CREATE OR REPLACE PUMP "STREAM_PUMP" AS 
      INSERT INTO "TEMP_STREAM"
         SELECT STREAM "heartRate", "rateType", ANOMALY_SCORE 
         FROM TABLE(RANDOM_CUT_FOREST(
                 CURSOR(SELECT STREAM * FROM "SOURCE_SQL_STREAM_001")));
   
   -- Sort records by descending anomaly score, insert into output stream
   CREATE OR REPLACE PUMP "OUTPUT_PUMP" AS 
      INSERT INTO "DESTINATION_SQL_STREAM"
         SELECT STREAM * FROM "TEMP_STREAM"
         ORDER BY FLOOR("TEMP_STREAM".ROWTIME TO SECOND), ANOMALY_SCORE DESC;

   O código lê linhas no SOURCE_SQL_STREAM_001, atribui uma pontuação de anomalias e grava as linhas resultantes em outro stream no aplicativo (TEMP_STREAM). O código de aplicativo, em seguida, classifica os registros no TEMP_STREAM e salva os resultados em outro stream no aplicativo (DESTINATION_SQL_STREAM). Você usa bombas para inserir linhas em streamings do aplicativo de entrada. Para obter mais informações, consulte Bombas e fluxos no aplicativo.

4. Configurar a saída: configure a saída de aplicativo para manter os dados no DESTINATION_SQL_STREAM para um destino externo, que é outro fluxo de dados do Kinesis. Verificar as pontuações de anomalias atribuídas a cada registro e determinar qual pontuação indica que uma anomalia (é preciso estar alerta) é externa ao aplicativo. Você pode usar uma AWS Lambda função para processar essas pontuações de anomalias e configurar alertas.

O exercício usa a região do Leste dos EUA (Norte da Virgínia) (us-east-1) para criar esses streams e o aplicativo. Se você usar qualquer outra região, precisará atualizar o código de acordo.

Tópicos

• Etapa 1: preparar

• Etapa 2: criar um aplicativo

• Etapa 3: Configuração da saída de aplicativo

• Etapa 4: verificar a saída

Próxima etapa

Etapa 1: preparar