Queryable State - Count within Time Window

[Velumani Duraisamy]

Hi,

I want to find the number of events happened so far in last 5 minutes and make that as a Queryable state. Is it possible? It will be of great help if someone provide some sample code for the same. 

Thanks,

Velu. 

[Fabian Hueske-2]

Hi,

you can implement that with a ProcessFunction [1].

The ProcessFunction would have to compute counts at some granularity (either event-time or processing-time) of records that are processed by the ProcessFunction in processElement().

I would do this with a MapState that has a timestamp as key and a count as value. The counts should be stored with some resolution, e.g., every 10 seconds (depends on your requirements).

This means you have one count for all elements that arrive within 10 seconds.

The actual count of the last 5 minutes is stored in a ValueState<Long>. and updated in regular intervals using timers.

When updating, you iterate over the MapState and discard all counts that are older than 5 minutes and compute the new count from the remaining counts.

The ValueState is configured to be queryable.

Best, Fabian

[1] https://ci.apache.org/projects/flink/flink-docs-release-1.4/dev/stream/operators/process_function.html

[2] https://ci.apache.org/projects/flink/flink-docs-release-1.4/dev/stream/state/queryable_state.html

2018-01-08 15:15 GMT+01:00 Velu Mitwa <[hidden email]>:

Hi,

I want to find the number of events happened so far in last 5 minutes and make that as a Queryable state. Is it possible? It will be of great help if someone provide some sample code for the same. 

Thanks,

Velu. 

[Velumani Duraisamy]

Thank you, Fabian, for the references. This is helpful.



--
Sent from: http://apache-flink-user-mailing-list-archive.2336050.n4.nabble.com/