Efficiently processing sparse events in a time windows

[Steven Murdoch]

Hello,

I am trying to do something that seems like it should be quite simple but I haven’t found an efficient way to do this with Flink and I expect I’m missing something obvious here.

The task is that I would like to process a sequence of events when a certain number appear within a keyed event-time window. There will be many keys but events within each keyed window will normally be quite sparse.

My first guess was to use Flink’s sliding windowing functionality. However my concern is that events are duplicated for each window. I would like to be precise about timing so every event would trigger hundreds of copies of an event in hundreds of windows, most which are then discarded because there are insufficient events.

My next guess was to use a process function, and maintain a queue of events as the state. When an event occurred I would add it to the queue and then remove any events which fell off the end of my window. I thought ListState would help here, but that appears to not allow items to be removed.

I then thought about using a ValueState with some queue data structure. However my understanding is that changes to a ValueState result in the entire object being copied and so would be quite inefficient and best avoided.

Finally I thought about trying to just maintain a series of timers – incrementing on an event and decrementing on its expiry. However I then hit the problem of timer coalescing. If an event occurs at the same time as its predecessor, the timer will not get set so the counter will get incremented but never decremented.

What I’m doing seems like it would be a common task but none of the options look good, so I feel I’m missing something. Could anyone offer some advice on how to handle this case?

Thanks in advance.

Best wishes,
Steven

[David Anderson-3]

Steven,

I'm pretty sure this is a scenario that doesn't have an obvious good solution. As you have discovered, the window API isn't much help; using a process function does make sense. The challenge is finding a data structure to use in keyed state that can be efficiently accessed and updated.

One option would be to use MapState, where the keys are timestamps (longs) and the values are lists of the events with the given timestamps (or just a count of those events, if that's sufficient). If you then use the RocksDB state backend, you can leverage an implementation detail of that state backend, which is that you can iterate over the entries in order, sorted by the key (the serialized, binary key), which in the case of keys that are longs, will do the right thing. Also, with the RocksDB state backend, you only have to do ser/de to access and update individual entries -- and not the entire map.

It's not exactly pretty to rely on this, and some of us have been giving some thought to adding a temporal state type to Flink that would make these scenarios feasible to implement efficiently on all of the state backends, but for now, this may be the best solution.

Regards,

David

On Wed, Sep 23, 2020 at 12:42 PM Steven Murdoch <[hidden email]> wrote:

Hello,

I am trying to do something that seems like it should be quite simple but I haven’t found an efficient way to do this with Flink and I expect I’m missing something obvious here.

The task is that I would like to process a sequence of events when a certain number appear within a keyed event-time window. There will be many keys but events within each keyed window will normally be quite sparse.

My first guess was to use Flink’s sliding windowing functionality. However my concern is that events are duplicated for each window. I would like to be precise about timing so every event would trigger hundreds of copies of an event in hundreds of windows, most which are then discarded because there are insufficient events.

My next guess was to use a process function, and maintain a queue of events as the state. When an event occurred I would add it to the queue and then remove any events which fell off the end of my window. I thought ListState would help here, but that appears to not allow items to be removed.

I then thought about using a ValueState with some queue data structure. However my understanding is that changes to a ValueState result in the entire object being copied and so would be quite inefficient and best avoided.

Finally I thought about trying to just maintain a series of timers – incrementing on an event and decrementing on its expiry. However I then hit the problem of timer coalescing. If an event occurs at the same time as its predecessor, the timer will not get set so the counter will get incremented but never decremented.

What I’m doing seems like it would be a common task but none of the options look good, so I feel I’m missing something. Could anyone offer some advice on how to handle this case?

Thanks in advance.

Best wishes,
Steven

[Steven Murdoch]

Thanks David, this is very helpful. I'm glad that it's not just that I had missed something obvious from the (generally very clear) documentation. I found various features that felt almost right (e.g. the priority queue behind Timers) but nothing that did the job. The temporal state idea does sound a very handy feature to have.

On Thu, 24 Sep 2020, at 08:50, David Anderson wrote:

Steven,

I'm pretty sure this is a scenario that doesn't have an obvious good solution. As you have discovered, the window API isn't much help; using a process function does make sense. The challenge is finding a data structure to use in keyed state that can be efficiently accessed and updated.

One option would be to use MapState, where the keys are timestamps (longs) and the values are lists of the events with the given timestamps (or just a count of those events, if that's sufficient). If you then use the RocksDB state backend, you can leverage an implementation detail of that state backend, which is that you can iterate over the entries in order, sorted by the key (the serialized, binary key), which in the case of keys that are longs, will do the right thing. Also, with the RocksDB state backend, you only have to do ser/de to access and update individual entries -- and not the entire map.

It's not exactly pretty to rely on this, and some of us have been giving some thought to adding a temporal state type to Flink that would make these scenarios feasible to implement efficiently on all of the state backends, but for now, this may be the best solution.

Regards,

David

On Wed, Sep 23, 2020 at 12:42 PM Steven Murdoch <[hidden email]> wrote:

Hello,

I am trying to do something that seems like it should be quite simple but I haven’t found an efficient way to do this with Flink and I expect I’m missing something obvious here.

The task is that I would like to process a sequence of events when a certain number appear within a keyed event-time window. There will be many keys but events within each keyed window will normally be quite sparse.

My first guess was to use Flink’s sliding windowing functionality. However my concern is that events are duplicated for each window. I would like to be precise about timing so every event would trigger hundreds of copies of an event in hundreds of windows, most which are then discarded because there are insufficient events.

My next guess was to use a process function, and maintain a queue of events as the state. When an event occurred I would add it to the queue and then remove any events which fell off the end of my window. I thought ListState would help here, but that appears to not allow items to be removed.

I then thought about using a ValueState with some queue data structure. However my understanding is that changes to a ValueState result in the entire object being copied and so would be quite inefficient and best avoided.

Finally I thought about trying to just maintain a series of timers – incrementing on an event and decrementing on its expiry. However I then hit the problem of timer coalescing. If an event occurs at the same time as its predecessor, the timer will not get set so the counter will get incremented but never decremented.

What I’m doing seems like it would be a common task but none of the options look good, so I feel I’m missing something. Could anyone offer some advice on how to handle this case?

Thanks in advance.

Best wishes,

Steven