How to detect babbling patterns using the flink cep library

Question

How to detect babbling patterns using the flink cep library ?

Example: Let say devices have some problem so, continuously it publishes values like on, off. How to detect patterns using CEP, If a problem exists for 30 mins. Some sample data I mentioned below.

OFF     16/08/18 11:38
ON      16/08/18 11:38
OFF     16/08/18 11:38
ON      16/08/18 11:37
OFF     16/08/18 11:37
ON      16/08/18 11:36
OFF     16/08/18 11:36
OFF     16/08/18 11:36
ON      16/08/18 11:36
OFF     16/08/18 11:35
ON      16/08/18 11:35
ON      16/08/18 11:34
OFF     16/08/18 11:34

I need to generate alerts for faulty devices. As you can see in above case faulty means every some sec duration device is sending ON or OFF continuously. — Avinash Tripathy
So what exactly is it that characterizes faulty devices? Is the event rate alone enough, or do you also need to consider that it is switching rapidly between OFF and ON? — David Anderson
I need to only consider rapidly switching devices ON and OFF for a particular duration of time. Let say if the device is rapidly switching to ON and OFF for 15 mins raise the alert. — Avinash Tripathy

David Anderson David Anderson · Accepted Answer · 2020-05-16T13:29:14

If your stream is in-order by time (it only matters that the stream is sorted for each individual device), then you could easily transform the stream to make this analysis easier. A RichFlatMapFunction like this will transform the sequence of ON OFF events into a sequence of state CHANGE events:

static class DetectChanges extends RichFlatMapFunction<String, String> {
    private transient ValueState<String> previousState;

    @Override
    public void open(Configuration parameters) throws Exception {
        previousState = getRuntimeContext().getState(new ValueStateDescriptor<>("previousState", String.class));
    }

    @Override
    public void flatMap(String onOrOff, Collector<String> out) throws Exception {

        if (previousState.value() != onOrOff) {
            out.collect("CHANGE");
            previousState.update(onOrOff);
        }
    }
}

Now the problem has been reduced to determining if the stream has some number of CHANGE events during an interval of time. This could easily be done with sliding windows, or you could use CEP if you like.

You could also do this entirely with CEP. Conceptually you might approach this as follows:

define an individual Pattern that matches ON+ OFF+
then define a Pattern group that matches that ON/OFF pattern whenever it occurs n times within some time interval

How to detect babbling patterns using the flink cep library

1 Answers