Ejemplo: Detección de anomalías de datos en una secuencia (función RANDOM_CUT_FOREST)

Amazon Kinesis Data Analytics proporciona una función RANDOM_CUT_FOREST () que puede asignar una puntuación de anomalía a cada registro en función de los valores de las columnas numéricas. Para obtener más información, consulte la RANDOM_CUT_FORESTreferencia SQL Function en Amazon Managed Service for Apache Flink.

En este ejercicio, escribirá el código de la aplicación para asignar una puntuación de anomalías a los registros en el origen de streaming de la aplicación. Haga lo siguiente para configurar la aplicación:

1. Configurar una fuente de transmisión: configura una transmisión de datos de Kinesis y escribe datos de muestra, como se muestra heartRate a continuación:

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

   El procedimiento proporciona un script de Python para que pueda rellenar la secuencia. Los valores de heartRate se generan de forma aleatoria. El 99 por ciento de los registros tienen valores de heartRate entre 60 y 100, y solo un 1 por ciento de ellos tienen valores de heartRate entre 150 y 200. Por lo tanto, los registros que tienen valores de heartRate comprendidos entre 150 y 200 son anomalías.

2. Configurar la entrada: mediante la consola, se crea una aplicación de Kinesis Data Analytics y se configura la entrada de la aplicación mediante la asignación de la fuente de transmisión a una transmisión SOURCE_SQL_STREAM_001 dentro de la aplicación (). Cuando se inicia la aplicación, Kinesis Data Analytics lee continuamente la fuente de transmisión e inserta los registros en la transmisión de la aplicación.

3. Especifique el código de la aplicación: el ejemplo utiliza el siguiente código de la aplicación:

   --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;

   El código lee filas de la puntuación SOURCE_SQL_STREAM_001, asigna un puntaje de anomalía, y escribe las filas resultante a otra secuencia en la aplicación (TEMP_STREAM). El código de la aplicación luego clasifica los registros en TEMP_STREAM y guarda los resultados en otra secuencia en la aplicación (DESTINATION_SQL_STREAM). Utilice bombas para insertar filas en secuencias en la aplicación. Para obtener más información, consulte Secuencias y bombeos en la aplicación.

4. Configurar la salida: configura la salida de la aplicación para conservar los datos en un destino externo, que es otra transmisión de datos de Kinesis. DESTINATION_SQL_STREAM La revisión de las puntuaciones de anomalías que se han asignado a cada registro y la determinación de qué puntuación indica que se ha producido una anomalía (y que es necesario que le avisen) es algo ajeno a la aplicación. Puede utilizar una función AWS Lambda para procesar estas puntuaciones de anomalías y configurar alertas.

El ejercicio utiliza el este de EE. UU. (Virginia del Norteus-east-1) () para crear estas transmisiones y su aplicación. Si utiliza cualquier otra región, debe actualizar el código en consecuencia.

Temas

• Paso 1: Preparación
• Paso 2: Cree una aplicación
• Paso 3: Configurar la salida de la aplicación
• Paso 4: Verificar la salida

Paso siguiente

Paso 1: Preparación