pre-initializing global window state

[jelmer]

Hi I am looking for some advice on how to solve the following problem

I'd like to keep track of the all time last n events received for a user.

An event on average takes up 500 bytes and here will be ten's of millions of users for which we need to keep this information. The list of events will be stored in cassandra for serving by an api

One way I can think of to implement this , is to use a global window per user with a count evictor. 

The problem I see with this is that the state would forever remain on the worker nodes, in our case, in rocks db.

This means there would be a *lot* of state to include for savepoints and checkpoints. This would make such a job very unwieldy to operate.

Is it possible to evict state from global state after some period of inactivity. and then reinitalize the global state with data loaded from cassandra when a new event arrives ? 

Or is there an obvious better way to tackle this problem that i am missing

Any pointers would be greatly appreciated

[Ken Krugler]

Hi Jelmer,

Three comments, if I understand your use case correctly…

1. I would first try using RockDB with incremental checkpointing, before deciding that an alternative approach is required.

2. Have you considered using queryable state vs. also keeping the list of events in Cassandra?

3. Depending on what you need the list of events for, often you can apply a streaming algorithm to get good-enough (approximate) results without storing complete state.

— Ken

On May 7, 2018, at 5:29 AM, jelmer <[hidden email]> wrote:

Hi I am looking for some advice on how to solve the following problem

I'd like to keep track of the all time last n events received for a user.

An event on average takes up 500 bytes and here will be ten's of millions of users for which we need to keep this information. The list of events will be stored in cassandra for serving by an api

One way I can think of to implement this , is to use a global window per user with a count evictor. 

The problem I see with this is that the state would forever remain on the worker nodes, in our case, in rocks db.

This means there would be a *lot* of state to include for savepoints and checkpoints. This would make such a job very unwieldy to operate.

Is it possible to evict state from global state after some period of inactivity. and then reinitalize the global state with data loaded from cassandra when a new event arrives ? 

Or is there an obvious better way to tackle this problem that i am missing

Any pointers would be greatly appreciated

--------------------------------------------

http://about.me/kkrugler

[phone]

[jelmer]

Hi Ken

> 1. I would first try using RockDB with incremental checkpointing, before deciding that an alternative approach is required.

That would reduce the size of the checkpoints but as far as I know not the savepoints. If i understand correctly that is still a copy of the entire system state. So deploying a new version of the application will be a daunting proposition involving  saving many gigabytes of data to external storage. and restarts that will take a very long time. Also our flink workers are not really scoped to these kind of storage requirements

> 2 Have you considered using queryable state vs. also keeping the list of events in Cassandra?

We looked at it before and at the time it was still somewhat experimental and somewhat immature with regards to handling failure scenarios. And it would require all the state to reside in flink. Which would again lead to long restarts when creating savepoints

> 3. Depending on what you need the list of events for, often you can apply a streaming algorithm to get good-enough (approximate) results without storing complete state.

Hyperloglog etc makes a lot of sense in many scenario's, but unfortunately this is not one of them. :-(

Another alternative I thought of for this problem is to  sidestep the window abstraction and fall back to a processing function with timers

On 7 May 2018 at 17:31, Ken Krugler <[hidden email]> wrote:

Hi Jelmer,

Three comments, if I understand your use case correctly…

1. I would first try using RockDB with incremental checkpointing, before deciding that an alternative approach is required.

2. Have you considered using queryable state vs. also keeping the list of events in Cassandra?

3. Depending on what you need the list of events for, often you can apply a streaming algorithm to get good-enough (approximate) results without storing complete state.

— Ken

On May 7, 2018, at 5:29 AM, jelmer <[hidden email]> wrote:

Hi I am looking for some advice on how to solve the following problem

I'd like to keep track of the all time last n events received for a user.

An event on average takes up 500 bytes and here will be ten's of millions of users for which we need to keep this information. The list of events will be stored in cassandra for serving by an api

One way I can think of to implement this , is to use a global window per user with a count evictor. 

The problem I see with this is that the state would forever remain on the worker nodes, in our case, in rocks db.

This means there would be a *lot* of state to include for savepoints and checkpoints. This would make such a job very unwieldy to operate.

Is it possible to evict state from global state after some period of inactivity. and then reinitalize the global state with data loaded from cassandra when a new event arrives ? 

Or is there an obvious better way to tackle this problem that i am missing

Any pointers would be greatly appreciated

--------------------------------------------

http://about.me/kkrugler

[phone]