How to have a true sliding window that ignores recent events?

Question

I was trying to build something like a window that behaves like a sliding window and:

• Counts events, ignoring the ones since the end of the window up to a certain "delay"
• Triggers once and only once per event
• Output count of events in [event TS - delay - duration , event TS - delay]
• Using pre-aggregation to avoid saving all the events.

The parameters of the window would be:

• Duration: duration of the window
• Output: offset of the events to trigger, counting from the end of the window. Analogous to "slide".
• Delay: offset of the events to ignore, counting from the end of the window. Essentially ignore events such that timestamp <= end of window - slide - delay.

The idea I was trying involved having a sliding window with:

• Duration: duration + output + delay
• Slide: output
• Trigger whenever the event TS is in [window end - output, window end]. This causes only one window to trigger.

The question now is: how to filter events in order to ignore the ones before "delay"? I've thought of:

• Having an aggregator that only sums the value if the event TS is between the correct bounds. This is not possible because aggregators in windows can't be a RichAggregateFunction and therefore I have no access to the window metadata. Is this assumption correct?

• Having pre-aggregation with:

  • Typical sum reducer
  • RichWindowFunction that uses managed state to keep track of how many elements were seen in the "area to ignore" and subtract that from the aggregator result received. The problem is that getRuntimeContext().getState() is not maintained per window and therefore can't be used. Is this assumption correct?

Are there any alternatives I'm missing or is any of the assumptions incorrect?

Answer 1

I may have gotten a bit lost in the details, but maybe I see a solution.

Seems like you could use a custom Trigger that fires twice, before and after the delay. Then use a ProcessWindowFunction with incremental aggregation, and use per-window state to hold the count of the first firing (and then subtract later).

Given the complexity in putting that all together, a solution based on a ProcessFunction and managed state might be simpler.