Ha, never mind, I realized I can just put the unique key into the aggregate object maintained by the FoldFunction.
I'm still curious why RichWindowFunction (and RichFoldFunction) aren't supported for Scala WindowedStream.apply. Mike On Thu, Mar 3, 2016 at 4:50 PM, Michael Radford <mub...@gmail.com> wrote: > 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. > > Thanks, > Mike > > On Wed, Mar 2, 2016 at 2:11 AM, Kostas Kloudas > <k.klou...@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 <aljos...@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 <mub...@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 >> >> >>
