Hi guys,
I'll try to come up with an example illustrating the behaviour over the
weekend.
Cheers,
Till
On Fri, Oct 14, 2016 at 11:16 AM, David Koch <ogd...@googlemail.com> wrote:
> Hello,
>
> Thanks for the code Sameer. Unfortunately, it didn't solve the issue.
> Compared to what I did the principle is the same - make sure that the
> watermark advances even without events present to trigger timeouts in CEP
> patterns.
>
> If Till or anyone else could provide a minimal example illustrating the
> supposed behaviour of:
>
> [CEP] timeout will be detected when the first watermark exceeding the
>> timeout value is received
>
>
> I'd very much appreciate it.
>
> Regards,
>
> David
>
>
> On Wed, Oct 12, 2016 at 1:54 AM, Sameer W <sam...@axiomine.com> wrote:
>
>> Try this. Your WM's need to move forward. Also don't use System
>> Timestamp. Use the timestamp of the element seen as the reference as the
>> elements are most likely lagging the system timestamp.
>>
>> DataStream<Event> withTimestampsAndWatermarks = tuples
>> .assignTimestampsAndWatermarks(new
>> AssignerWithPeriodicWatermarks<Event>()
>> {
>>
>> long waterMarkTmst;
>> long lastEmittedWM=0;
>> @Override
>> public long extractTimestamp(Event element, long
>> previousElementTimestamp) {
>> if(element.tmst>lastEmittedWM){
>> waterMarkTmst = element.tmst-1; //Assumes increasing
>> timestamps. Need to subtract 1 as more elements with same TS might arrive
>> }
>> return element.tmst;
>> }
>>
>> @Override
>> public Watermark getCurrentWatermark() {
>> if(lastEmittedWM==waterMarkTmst){ //No new event seen,
>> move the WM forward by auto watermark interval
>> waterMarkTmst = waterMarkTmst + 1000l//Increase by
>> auto watermark interval (Watermarks only move forward in time)
>> }
>> lastEmittedWM = waterMarkTmst
>>
>> System.out.println(String.format("Watermark at %s", new
>> Date(waterMarkTmst)));
>> return new Watermark(waterMarkTmst);//Until an event is
>> seem WM==0 starts advancing by 1000ms until an event is seen
>> }
>> }).keyBy("key");
>>
>> On Tue, Oct 11, 2016 at 7:29 PM, David Koch <ogd...@googlemail.com>
>> wrote:
>>
>>> Hello,
>>>
>>> I tried setting the watermark to System.currentTimeMillis() - 5000L,
>>> event timestamps are System.currentTimeMillis(). I do not observe the
>>> expected behaviour of the PatternTimeoutFunction firing once the watermark
>>> moves past the timeout "anchored" by a pattern match.
>>>
>>> Here is the complete test class source <http://pastebin.com/9WxGq2wv>,
>>> in case someone is interested. The timestamp/watermark assigner looks like
>>> this:
>>>
>>> DataStream<Event> withTimestampsAndWatermarks = tuples
>>> .assignTimestampsAndWatermarks(new
>>> AssignerWithPeriodicWatermarks<Event>() {
>>>
>>> long waterMarkTmst;
>>>
>>> @Override
>>> public long extractTimestamp(Event element, long
>>> previousElementTimestamp) {
>>> return element.tmst;
>>> }
>>>
>>> @Override
>>> public Watermark getCurrentWatermark() {
>>> waterMarkTmst = System.currentTimeMillis() - 5000L;
>>> System.out.println(String.format("Watermark at %s", new
>>> Date(waterMarkTmst)));
>>> return new Watermark(waterMarkTmst);
>>> }
>>> }).keyBy("key");
>>>
>>> withTimestampsAndWatermarks.getExecutionConfig().setAutoWate
>>> rmarkInterval(1000L);
>>>
>>> // Apply pattern filtering on stream.
>>> PatternStream<Event> patternStream =
>>> CEP.pattern(withTimestampsAndWatermarks,
>>> pattern);
>>>
>>> Any idea what's wrong?
>>>
>>> David
>>>
>>>
>>> On Tue, Oct 11, 2016 at 10:20 PM, Sameer W <sam...@axiomine.com> wrote:
>>>
>>>> Assuming an element with timestamp which is later than the last emitted
>>>> watermark arrives, would it just be dropped because the PatternStream does
>>>> not have a max allowed lateness method? In that case it appears that CEP
>>>> cannot handle late events yet out of the box.
>>>>
>>>> If we do want to support late events can we chain a
>>>> keyBy().timeWindow().allowedLateness(x).map().assignTimestampsAndWatermarks().keyBy()
>>>> again before handing it to the CEP operator. This way we may have the
>>>> patterns fired multiple times but it allows an event to be late and out of
>>>> order. It looks like it will work but is there a less convoluted way.
>>>>
>>>> Thanks,
>>>> Sameer
>>>>
>>>> On Tue, Oct 11, 2016 at 12:17 PM, Till Rohrmann <
>>>> till.rohrm...@gmail.com> wrote:
>>>>
>>>>> But then no element later than the last emitted watermark must be
>>>>> issued by the sources. If that is the case, then this solution should
>>>>> work.
>>>>>
>>>>> Cheers,
>>>>> Till
>>>>>
>>>>> On Tue, Oct 11, 2016 at 4:50 PM, Sameer W <sam...@axiomine.com> wrote:
>>>>>
>>>>>> Hi,
>>>>>>
>>>>>> If you know that the events are arriving in order and a consistent
>>>>>> lag, why not just increment the watermark time every time the
>>>>>> getCurrentWatermark() method is invoked based on the
>>>>>> autoWatermarkInterval
>>>>>> (or less to be conservative).
>>>>>>
>>>>>> You can check if the watermark has changed since the arrival of the
>>>>>> last event and if not increment it in the getCurrentWatermark() method.
>>>>>> Otherwise the watermark will never increase until an element arrive and
>>>>>> if
>>>>>> the stream partition stalls for some reason the whole pipeline freezes.
>>>>>>
>>>>>> Sameer
>>>>>>
>>>>>>
>>>>>> On Tue, Oct 11, 2016 at 6:04 AM, Till Rohrmann <
>>>>>> till.rohrm...@gmail.com> wrote:
>>>>>>
>>>>>>> Hi David,
>>>>>>>
>>>>>>> the problem is still that there is no corresponding watermark saying
>>>>>>> that 4 seconds have now passed. With your code, watermarks will be
>>>>>>> periodically emitted but the same watermark will be emitted until a new
>>>>>>> element arrives which will reset the watermark. Thus, the system can
>>>>>>> never
>>>>>>> know until this watermark is seen whether there will be an earlier
>>>>>>> event or
>>>>>>> not. I fear that this is a fundamental problem with stream processing.
>>>>>>>
>>>>>>> You're right that the negation operator won't solve the problem. It
>>>>>>> will indeed suffer from the same problem.
>>>>>>>
>>>>>>> Cheers,
>>>>>>> Till
>>>>>>>
>>>>>>> On Sun, Oct 9, 2016 at 7:37 PM, <lg...@yahoo.com> wrote:
>>>>>>>
>>>>>>>> >>FLINK-3320 <https://issues.apache.org/jira/browse/FLINK-3320> (CEP
>>>>>>>> "not" operator) does not address this because again, how would the "not
>>>>>>>> match" be triggered if no event at all occurs?
>>>>>>>>
>>>>>>>> Good question.
>>>>>>>>
>>>>>>>> I'm not sure whether the following will work:
>>>>>>>>
>>>>>>>> This could be done by creating a CEP matching pattern that uses
>>>>>>>> both of "notNext" (or "notFollowedBy") and "within" constructs.
>>>>>>>> Something
>>>>>>>> like this:
>>>>>>>>
>>>>>>>> Pattern<Event, ?> pattern = Pattern.<Event>begin("first")
>>>>>>>> .notNext("second")
>>>>>>>> .within(Time.seconds(3));
>>>>>>>>
>>>>>>>> I'm hoping Flink CEP experts (Till?) will comment on this.
>>>>>>>>
>>>>>>>> Note: I have requested these negation patterns to be implemented in
>>>>>>>> Flink CEP, but notNext/notFollowedBy are not yet implemented in Flink..
>>>>>>>>
>>>>>>>>
>>>>>>>> - LF
>>>>>>>>
>>>>>>>>
>>>>>>>>
>>>>>>>>
>>>>>>>> ------------------------------
>>>>>>>> *From:* David Koch <ogd...@googlemail.com>
>>>>>>>> *To:* user@flink.apache.org; lg...@yahoo.com
>>>>>>>> *Sent:* Sunday, October 9, 2016 5:51 AM
>>>>>>>>
>>>>>>>> *Subject:* Re: Listening to timed-out patterns in Flink CEP
>>>>>>>>
>>>>>>>> Hello,
>>>>>>>>
>>>>>>>> Thank you for the explanation as well as the link to the other
>>>>>>>> post. Interesting to learn about some of the open JIRAs.
>>>>>>>>
>>>>>>>> Indeed, I was not using event time, but processing time. However,
>>>>>>>> even when using event time I only get notified of timeouts upon
>>>>>>>> subsequent
>>>>>>>> events.
>>>>>>>>
>>>>>>>> The link <http://pastebin.com/x4m3RHQz> contains an example where
>>>>>>>> I read <key> <value> from a socket, wrap this in a custom "event" with
>>>>>>>> timestamp, key the resultant stream by <key> and attempt to detect
>>>>>>>> <key>
>>>>>>>> instances no further than 3 seconds apart using CEP.
>>>>>>>>
>>>>>>>> Apart from the fact that results are only printed when I close the
>>>>>>>> socket (normal?) I don't observe any change in behaviour
>>>>>>>>
>>>>>>>> So event-time/watermarks or not: SOME event has to occur for the
>>>>>>>> timeout to be triggered.
>>>>>>>>
>>>>>>>> FLINK-3320 <https://issues.apache.org/jira/browse/FLINK-3320> (CEP
>>>>>>>> "not" operator) does not address this because again, how would the "not
>>>>>>>> match" be triggered if no event at all occurs?
>>>>>>>>
>>>>>>>> On Sat, Oct 8, 2016 at 12:50 AM, <lg...@yahoo.com> wrote:
>>>>>>>>
>>>>>>>> The following is a better link:
>>>>>>>>
>>>>>>>> http://mail-archives.apache. org/mod_mbox/flink-user/
>>>>>>>> 201609.mbox/%3CCAC27z% 3DOTtv7USYUm82bE43- DkoGfVC4UAWD6uQwwRgTsE5be8g%
>>>>>>>> 40mail.gmail.com%3E
>>>>>>>> <http://mail-archives.apache.org/mod_mbox/flink-user/201609.mbox/%3CCAC27z%3DOTtv7USYUm82bE43-DkoGfVC4UAWD6uQwwRgTsE5be8g%40mail.gmail.com%3E>
>>>>>>>>
>>>>>>>>
>>>>>>>> - LF
>>>>>>>>
>>>>>>>>
>>>>>>>>
>>>>>>>>
>>>>>>>> ------------------------------
>>>>>>>> *From:* "lg...@yahoo.com" <lg...@yahoo.com>
>>>>>>>> *To:* "user@flink.apache.org" <user@flink.apache.org>
>>>>>>>> *Sent:* Friday, October 7, 2016 3:36 PM
>>>>>>>>
>>>>>>>> *Subject:* Re: Listening to timed-out patterns in Flink CEP
>>>>>>>>
>>>>>>>> Isn't the upcoming CEP negation (absence of an event) feature
>>>>>>>> solve this issue?
>>>>>>>>
>>>>>>>> See this discussion thread:
>>>>>>>> http://mail-archives.apache. org/mod_mbox/flink-user/
>>>>>>>> 201609.mbox/%3CCAC27z%3DOD% 2BTq8twBw_ 1YKni5sWAU3g1S9WDpJw0DUwgiG9YX
>>>>>>>> 9Fg%40mail.gmail.com%3E
>>>>>>>> <http://mail-archives.apache.org/mod_mbox/flink-user/201609.mbox/%3CCAC27z%3DOD%2BTq8twBw_1YKni5sWAU3g1S9WDpJw0DUwgiG9YX9Fg%40mail.gmail.com%3E>
>>>>>>>>
>>>>>>>>
>>>>>>>>
>>>>>>>> // Atul
>>>>>>>>
>>>>>>>>
>>>>>>>> ------------------------------
>>>>>>>> *From:* Till Rohrmann <trohrm...@apache.org>
>>>>>>>> *To:* user@flink.apache.org
>>>>>>>> *Sent:* Friday, October 7, 2016 12:58 AM
>>>>>>>> *Subject:* Re: Listening to timed-out patterns in Flink CEP
>>>>>>>>
>>>>>>>> Hi David,
>>>>>>>>
>>>>>>>> in case of event time, the timeout will be detected when the first
>>>>>>>> watermark exceeding the timeout value is received. Thus, it depends a
>>>>>>>> little bit how you generate watermarks (e.g. periodically, watermark
>>>>>>>> per
>>>>>>>> event).
>>>>>>>>
>>>>>>>> In case of processing time, the time is only updated whenever a new
>>>>>>>> element arrives. Thus, if you have an element arriving 4 seconds after
>>>>>>>> Event A, it should detect the timeout. If the next event arrives 20
>>>>>>>> seconds
>>>>>>>> later, than you won't see the timeout until then.
>>>>>>>>
>>>>>>>> In the case of processing time, we could think about registering
>>>>>>>> timeout timers for processing time. However, I would highly recommend
>>>>>>>> you
>>>>>>>> to use event time, because with processing time, Flink cannot guarantee
>>>>>>>> meaningful computations, because the events might arrive out of order.
>>>>>>>>
>>>>>>>> Cheers,
>>>>>>>> Till
>>>>>>>>
>>>>>>>> On Thu, Oct 6, 2016 at 3:08 PM, David Koch <ogd...@googlemail.com>
>>>>>>>> wrote:
>>>>>>>>
>>>>>>>> Hello,
>>>>>>>>
>>>>>>>> With Flink CEP, is there a way to actively listen to pattern
>>>>>>>> matches that time out? I am under the impression that this is not
>>>>>>>> possible.
>>>>>>>>
>>>>>>>> In my case I partition a stream containing user web navigation by
>>>>>>>> "userId" to look for sequences of Event A, followed by B within 4
>>>>>>>> seconds
>>>>>>>> for each user.
>>>>>>>>
>>>>>>>> I registered a PatternTimeoutFunction which assuming a non-match
>>>>>>>> only fires upon the first event after the specified timeout. For
>>>>>>>> example,
>>>>>>>> given user X: Event A, 20 seconds later Event B (or any other type of
>>>>>>>> event).
>>>>>>>>
>>>>>>>> I'd rather have a notification fire directly upon the 4 second
>>>>>>>> interval expiring since passive invalidation is not really applicable
>>>>>>>> in my
>>>>>>>> case.
>>>>>>>>
>>>>>>>> How, if at all can this be achieved with Flink CEP?
>>>>>>>>
>>>>>>>> Thanks,
>>>>>>>>
>>>>>>>> David
>>>>>>>>
>>>>>>>>
>>>>>>>>
>>>>>>>>
>>>>>>>>
>>>>>>>>
>>>>>>>>
>>>>>>>>
>>>>>>>>
>>>>>>>>
>>>>>>>
>>>>>>
>>>>>
>>>>
>>>
>>
>
