Re: CEP condition expression and its event consuming strategy

[Chao Wang]

Hi Dawid,

Thank you.
Ad. 1 I noticed that the method getEventsForPattern() returns an 
Iterable<T> and we need to further invoke .operator().next() to get 
access to the event value.

Ad. 2 Here is a bit about a use case we have that calls for such 
discarding semantics. In the event processing project I am currently 
working on, input event streams are sensor data, and we join streams and 
do Kalman filtering, FFT, etc. We therefore choose to discard the 
accepted events once the data they carry have been processed; otherwise, 
it may cause duplicated processing as well as incorrect join semantics.

We came up with this question while doing an empirical comparison of 
Flink and our system (implemented with the TAO real-time event service). 
We implemented in our system such semantics, by removing input events 
once CEP emits the corresponding output events.

Could you provide some use cases where the discarding semantics are not 
needed? I guess I am wired into processing sensor data and thus cannot 
think of a case where reusing accepted events would be of interest. 
Also, could you share some pointers to streaming application in-use? We 
are seeking to make our research work more relevant to current practice.

Thank you very much,

Chao

On 07/27/2017 02:17 AM, Dawid Wysakowicz wrote:
Hi Chao,

Ad. 1 You could implement it with IterativeCondition. Sth like this:

Pattern<Event, ?> pattern = Pattern.<Event>begin("first").where(new 
SimpleCondition<Event>() {
    @Override
    public boolean filter(Event value) throws Exception {
       return value.equals("A") || value.equals("B");
    }
}).followedBy("second").where(new IterativeCondition<Event>() {
    @Override
    public boolean filter(Event value, Context<Event> ctx) throws Exception {
       return (value.equals("A") || value.equals("B")) && 
!value.equals(ctx.getEventsForPattern("first"));
    }
}).

Ad. 2 Unfortunately right now as you said Pattern restarts each other event and 
it is not possible to change that strategy. There is ongoing work to introduce 
AfterMatchSkipStrategy[1], but at best it will be merged in 1.4.0. I did not 
give it much thought, but I would try implement some discarding logic.

Regards,
Dawid

[1] https://issues.apache.org/jira/browse/FLINK-7169

On 26 Jul 2017, at 22:45, Chao Wang <chaow...@wustl.edu> wrote:

Hi,

I have two questions regarding the use of the Flink CEP library 
(flink-cep_2.11:1.3.1), as follows:

1. I'd like to know how to use the API to express "emit event C in the presence of events A and B, with 
no restriction on the arriving order of A and B"? I've tried by creating two patterns, one for "A 
and then B" and the other for "B and then A", and consequently using two patternStreams to 
handle each case, which emits C. It worked but to me this approach seems redundant.

2. Given the above objective expression, how to consume the accepted events so 
that they will not be used for future matchings? For example, with the arriving 
sequence {A, B, A}, the CEP should only emit one C (due to the matching of 
{A,B}), not two Cs (due to {A,B} and {B,A}). Similarly, with the arriving 
sequence {B, A, B, A}, the CPE should only emit two Cs, not three.


Thanks,

Chao