Maintaining watermarks per key, instead of per operator instance

[Aljoscha Krettek]

I think you would need something like this:

var hourlyDiscarding = stream

  .window(1.hour)

  .trigger(discarding)

  .apply(..)

//write to cassandra

hourlyDiscarding

  .window(1.hour)

  .trigger(accumulating)

  .apply(..)

  .addSink(cassandra)

//forward to next acc step

var daily = hourlyDiscarding

  .window(1.day)

  .trigger(accumulating)

  .apply(…)

//write to cassandra

daily.addSink(cassandra)

The decision between accumulating/discarding happens at the point where the window is defined, not downstream (this is the same as in Beam).

On Wed, 23 Nov 2016 at 11:37 kaelumania <[hidden email]> wrote:

Sounds good to me. But I still need to have some kind of side output (cassandra) that stores the accumulating aggregates on each time scale (minute, hour). Thus I would need to have something like this

var hourly = stream.window(1.hour).apply(..)

//write to cassandra

hourly.trigger(accumulating).addSink(cassandra)

//forward to next acc step

var daily = hourly.trigger(discarding).window(1.day).apply(…)

//write to cassandra

daily.trigger(accumulating).addSink(cassandra)

Would this be possible?

best, Stephan

On 23 Nov 2016, at 11:16, Aljoscha Krettek [via Apache Flink User Mailing List archive.] <[hidden email]> wrote:

You can implement discarding behaviour by writing a custom trigger (based on EventTimeTrigger) that returns FIRE_AND_PURGE when firing. With this you could maybe implement a cascade of windows where the first aggregates for the smallest time interval and is discarding and where the other triggers take these "pre-aggregated" values and accumulate.

On Tue, 22 Nov 2016 at 08:11 Stephan Epping <<a href="x-msg://8/user/SendEmail.jtp?type=node&amp;node=10294&amp;i=0" target="_top" rel="nofollow" link="external" class="">[hidden email]> wrote:

Hey Aljoscha,

the first solution did not work out as expected. As when late elements arrive the first window is triggered again and would emit a new (accumulated) event, that would be counted twice (in time accumulation and late accumulation) in the second window.I could implement my own (discarding strategy) like in Apache Beam, but the out stream should contain accumulated events that are stored in cassandra. The second solution just gave an compiler error, thus I think is not possible right now.

best Stephan