flink-user mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From Michael Radford <mub...@gmail.com>
Subject Re: Windows, watermarks, and late data
Date Fri, 04 Mar 2016 00:50:20 GMT
Thank you, that was helpful. I didn't appreciate that a Trigger is
fully in control of when to fire / purge regardless of the watermark.

Now I am wondering the best way to distinguish different instances of
the same time window with completely different data, vs. repeated
fires that include data used in previous fires. More specifically:

- If the data is not late, I will hold onto an aggregate maintained by
a FoldFunction, and fire the window periodically without purging. Each
of these fires should carry some key that is the same for every fire,
so that a database update downstream can overwrite the previous
partial aggregates.

- If the data is late, I will only hold onto the aggregate for some
period, and then fire and purge. But this late fire should carry a
different key, so that  the aggregates from the non-late data are not
overwritten. That way, I am able to deal with arbitrarily late data
statelessly without knowing what aggregates for the time window have
already been written.

It looks like I could do this by using RichWindowFunction in the Java
API, and saving the key using the RuntimeContext state API. However, I
can't seem to pass a RichWindowFunction to the Scala WindowedStream's
apply method. Is there any easy way around this?

I was also hoping that the FoldFunction passed to WindowedStream.apply
could be a RichFoldFunction, but that is specifically prohibited for
some reason.

Any hints on how to make a stateful WindowedStream.apply in Scala
would be much appreciated.


On Wed, Mar 2, 2016 at 2:11 AM, Kostas Kloudas
<k.kloudas@data-artisans.com> wrote:
> Hello Mike,
> The code that Aljiosha mentioned is here:
> https://github.com/kl0u/flink-examples/blob/master/src/main/java/com/dataartisans/flinksolo/customTriggers/EventTimeTriggerWithEarlyAndLateFiring.java
> This allows you to specify a trigger like:
> EventTimeTriggerWithEarlyAndLateFiring trigger =
> EventTimeTriggerWithEarlyAndLateFiring.create()
> .withEarlyFiringEvery(Time.minutes(10))
> .withLateFiringEvery(Time.minutes(5))
> .withAllowedLateness(Time.minutes(20))
> .accumulating();
> The means that it will fire every 10 minutes (in processing time) until the
> end of the window (event time), and then
> every 5 minutes (processing time) for late elements up to 20 minutes late.
> In addition, previous elements are not discarded.
> Hope this helps,
> Kostas
> On Mar 2, 2016, at 11:02 AM, Aljoscha Krettek <aljoscha@apache.org> wrote:
> Hi,
> I did some initial work on extending the EventTimeTrigger a bit to allow
> more complex behavior. Specifically, this allows setting an “allowed
> lateness” after which elements should no longer lead to windows being
> emitted. Also, it allows to specify to keep an emitted window in memory and
> when a late element arrives emit the whole window again.
> The code I have is here:
> https://github.com/aljoscha/flink/blob/window-late/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/windowing/triggers/EventTimeTrigger.java
> Kostas Kloudas worked on extending it, so maybe he could share his version
> of the trigger as well.
> Cheers,
> Aljoscha
> On 01 Mar 2016, at 18:35, Michael Radford <mubber@gmail.com> wrote:
> I'm evaluating Flink for a reporting application that will keep
> various aggregates updated in a database. It will be consuming from
> Kafka queues that are replicated from remote data centers, so in case
> there is a long outage in replication, I need to decide what to do
> about windowing and late data.
> If I use Flink's built-in windows and watermarks, any late data will
> be come in 1-element windows, which could overwhelm the database if a
> large batch of late data comes in and they are each mapped to
> individual database updates.
> As far as I can tell, I have two options:
> 1. Ignore late data, by marking it as late in an
> AssignerWithPunctuatedWatermarks function, and then discarding it in a
> flatMap operator. In this scenario, I would rely on a batch process to
> fill in the missing data later, in the lambda architecture style.
> 2. Implement my own watermark logic to allow full windows of late
> data. It seems like I could, for example, emit a "tick" message that
> is replicated to all partitions every n messages, and then a custom
> Trigger could decide when to purge each window based on the ticks and
> a timeout duration. The system would never emit a real Watermark.
> My questions are:
> - Am I mistaken about either of these, or are there any other options
> I'm not seeing for avoiding 1-element windows?
> - For option 2, are there any problems with not emitting actual
> watermarks, as long as the windows are eventually purged by a trigger?
> Thanks,
> Mike

View raw message