flink-user mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From Chao Wang <chaow...@wustl.edu>
Subject Re: CEP condition expression and its event consuming strategy
Date Thu, 03 Aug 2017 18:57:10 GMT
Thank you, Dawid. FYI, I've implemented the discarding logic by 
CoFlatMapFunction, for the special case where there are only two input 
streams: I maintain a logical state (no match, input1 matched, or input2 
matched) and use private variables to store the matched event so far, 
which waits to be processed along with the event from the other input 


On 07/31/2017 02:13 AM, Dawid Wysakowicz wrote:
> Ad. 1 Yes it returns and Iterable to support times and oneOrMore patterns(which can accept
more than one event).
> Ad. 2 Some use case for not discarding used events could be e.g. looking for some shapes
in our data, e.g. W-shapes. In this case one W-shape could start on the middle peak of the
previous one.
> Unfortunately personally I can’t point you to any in-use applications. Maybe Kostas,
I’ve added to the discussion, know of any.
> Anyway, thanks for interest in the CEP library. We will be happy to hear any comments
and suggestions for future improvements.
>> On 28 Jul 2017, at 21:54, Chao Wang <chaowang@wustl.edu> wrote:
>> 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 <chaowang@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
>>>> 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

View raw message