flink-user mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From Aljoscha Krettek <aljos...@apache.org>
Subject Re: Firing windows multiple times
Date Mon, 29 Aug 2016 08:56:14 GMT
that would certainly be possible? What do you think can be gained by having
knowledge about the current watermark in the WindowFunction, in a specific
case, possibly?


On Wed, 24 Aug 2016 at 23:21 Shannon Carey <scarey@expedia.com> wrote:

> What do you think about adding the current watermark to the window
> function metadata in FLIP-2?
> From: Shannon Carey <scarey@expedia.com>
> Date: Friday, August 12, 2016 at 6:24 PM
> To: Aljoscha Krettek <aljoscha@apache.org>, "user@flink.apache.org" <
> user@flink.apache.org>
> Subject: Re: Firing windows multiple times
> Thanks Aljoscha, I didn't know about those. Yes, they look like handy
> changes, especially to enable flexible approaches for eviction. In
> particular, having the current watermark available to the evictor via
> EvictorContext is helpful: it will be able to evict the old data more
> easily without needing to rely on Window#maxTimestamp().
> However, I think you might still be missing a piece. Specifically, it
> would still not be possible for the window function to choose which items
> to aggregate based on the current watermark. In particular, it is desirable
> to be able to aggregate only the items below the watermark, omitting items
> which have come in with timestamps larger than the watermark. Does that
> make sense?
> -Shannon
> From: Aljoscha Krettek <aljoscha@apache.org>
> Date: Friday, August 12, 2016 at 4:25 AM
> To: "user@flink.apache.org" <user@flink.apache.org>
> Subject: Re: Firing windows multiple times
> Hi,
> there is already this FLIP:
> https://cwiki.apache.org/confluence/display/FLINK/FLIP-4+%3A+Enhance+Window+Evictor which
> also links to a mailing list discussion. And this FLIP:
> https://cwiki.apache.org/confluence/display/FLINK/FLIP-2+Extending+Window+Function+Metadata.
> The former proposes to enhance the Evictor API a bit, among other things we
> propose to give the evictor access to the current watermark. The other FLIP
> proposes to extend the amount of meta-data we give to the window function.
> The first to things we propose to add is a "firing reason" that would tell
> you whether this was an early firing, an on time firing or a late firing.
> The second thing is a firing counter that would tell you how many times the
> trigger has fired so far for the current window.
> Would a combination of these help with your use case?
> Cheers,
> Aljoscha
> On Thu, 11 Aug 2016 at 19:19 Shannon Carey <scarey@expedia.com> wrote:
>> "If Window B is a Folding Window and does not have an evictor then it
>> should not keep the list of all received elements."
>> Agreed! Upon closer inspection, the behavior I'm describing is only
>> present when using EvictingWindowOperator, not when using WindowOperator. I
>> misread line 382 of WindowOperator which calls windowState.add(): in
>> actuality, the windowState is a FoldingState which incorporates the
>> user-provided fold function in order to eagerly fold the data. In contrast,
>> if you use an evictor, EvictingWindowOperator has the behavior I describe.
>> I am already using a custom Trigger which uses a processing timer to FIRE
>> a short time after a new event comes in, and an event timer to
>> It seems that I can achieve the desired effect by avoiding use of an
>> evictor so that the intermediate events are not retained in an
>> EvictingWindowOperator's state, and perform any necessary eviction within
>> my fold function. This has the aforementioned drawbacks of the windowed
>> fold function not knowing about watermarks, and therefore it is difficult
>> to be precise about choosing which items to evict. However, this seems to
>> be the best choice within the current framework.
>> Interestingly, it appears that TimeEvictor doesn't really know about
>> watermarks either. When a window emits an event, regardless of how it was
>> fired, it is assigned the timestamp given by its window's maxTimestamp(),
>> which might be much greater than the processing time that actually fired
>> the event. Then, TimeEvictor compares the max timestamp of all items in the
>> window against the other ones in order to determine which ones to evict.
>> Basically, it assumes that the events were emitted due to the window
>> terminating with FIRE_AND_PURGE. What if we gave more information
>> (specifically, the current watermark) to the evictor in order to allow it
>> to deal with a mix of intermediate events (fired by processing time) and
>> final events (fired by event time when the watermark reaches the window)?
>> That value is already available in the WindowOperator & could be passed to
>> the Evictor very easily. It would be an API change, of course.
>> Other than that, is it worth considering a change to
>> EvictingWindowOperator to allow user-supplied functions to reduce the size
>> of its state when people fire upstream windows repeatedly? From what I see
>> when I monitor the state with debugger print statements, the
>> EvictingWindowOperator is definitely holding on to all the elements ever
>> received, not just the aggregated result. You can see this clearly because
>> EvictingWindowOperator holds a ListState instead of a FoldingState. The
>> user-provided fold function is only applied upon fire().
>> -Shannon

View raw message