wso2 - WS02 CEP Consultas Siddhi

wso2cep

Question

Nuevo en Siddhi CEP. Aparte de los documentos regulares en WS02 CEP alguien puede apuntar a un buen tutorial.

Aquí están nuestros requisitos. Señale algunas pistas sobre las formas correctas de escribir tales consultas.

• Tener un solo flujo de notificación del dispositivo sensor (aplicación IOT).
• La entrada de flujo se realiza a través de REST-JSON y la salida también se formateará en REST-JSON. (Espero que esto sea posible en WS02 CEP 3.1)

Tipo de plan de ejecución requerido: - Si la notificación del dispositivo informa el uso del Sensor 1, luego monitoree para ver si dentro de 5 minutos si la notificación del dispositivo informa el uso del Sensor 2 también. Si se encuentra, genere flujo de salida informando actividad compuesta en REST-JSON.
- Si dicha actividad compuesta no se detecta durante un intervalo de tiempo en la mañana, tarde y noche, genere el estado de advertencia de evento de eventos en REST-JSON. (Entonces, ¿cómo encontrar eventos que no ocurrieron a tiempo? - Si tal actividad compuesta no se encuentra dentro de algunos intervalos de tiempo en la mañana, tarde y noche, informe el estado de fail1-event-stream en REST-JSON.

Esto debería funcionar todos los días, entonces, ¿cómo se eliminarán los datos procesados ​​previos en WSO2 CEP?

Saludos, Amit

Answer 1

Las consultas pueden ser las siguientes (estas son consultas preliminares y pueden requerir ligeras modificaciones para que se ejecuten)

1. Para detectar el sensor 1, y luego el sensor 2 en 5 minutos (suponiendo que sensorStram tiene id, valor) simplemente puede usar un patrón como el siguiente con la palabra clave ''inside'':

from e1=sensorStream[sensorId == ''1''] -> e2=sensorStream[sensorId == ''2'']

select ''composite activity detected'' as description, e1.value as sensor1Value, e2.value as sensor2Value

within 5 minutes

insert into compositeActivityStream;

2. Para detectar no ocurrencias (id = 1 llega, pero no id = 2 en 5 minutos) podemos tener dos consultas siguientes:

from sensorStream[sensorId == ''1'']#window.time(5 minutes)

select *

insert into delayedSensor1Stream for expired-events;


from e1=sensorStream[sensorId == ''1''] -> nonOccurringEvent = sensorStream[sensorId == ''2''] or delayedEvent=delayedSensor1Stream

select ''id=2 not found'' as description, e1.value as id1Value, nonOccurringEvent.sensorId as nonOccurringId

having (not(nonOccurringId instanceof string))

insert into nonOccurrenceStream;

Esto detectará no ocurrencias inmediatamente al final de 5 minutos después de la llegada de id = 1 evento. Para una explicación de la lógica anterior, eche un vistazo a la muestra no ocurrente de cep 4.0.0 (la sintaxis es un poco diferente, pero la misma idea)

3. Ahora que necesita generar periódicamente un informe, necesitamos otra consulta. Por comodidad, supongo que necesita un informe cada 6 horas (360 minutos) y utiliza una ventana de lote de tiempo aquí. Alternativamente, con el nuevo CEP 4.0.0 puede usar la ''ventana de Cron'' para generar esto en momentos específicos que es mejor para su caso de uso.

from nonOccurrenceStream#window.timeBatch(360 minutes)

select count(id1Value) as nonOccurrenceCount

insert into nonOccurrenceReportsStream for expired-events;

Puede usar adaptadores de entrada / salida http y hacer asignaciones json con constructores y formateadores json para este caso de uso.