Queryable State - Count within Time Window

[Velu Mitwa <ve...@gmail.com>]

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.

Re: Queryable State - Count within Time Window

[Velumani Duraisamy <ve...@gmail.com>]

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



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

Re: Queryable State - Count within Time Window

[Fabian Hueske <fh...@gmail.com>]

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 <ve...@gmail.com>:

> 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.
>
>