flink-user mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From Hequn Cheng <chenghe...@gmail.com>
Subject Re: Question regarding state cleaning using timer
Date Mon, 17 Sep 2018 02:00:12 GMT
Hi bhaskar,

You need change nothing if you want to handle multi keys. Flink will do it
for you. The ValueState is a keyed state. You can think of Keyed State
<https://ci.apache.org/projects/flink/flink-docs-stable/dev/stream/state/state.html#using-managed-keyed-state>
as Operator State that has been partitioned, or sharded, with exactly one
state-partition per key.
TTL can be used in the same way.

Best, Hequn


On Fri, Sep 14, 2018 at 10:29 PM bhaskar.ebay77@gmail.com <
bhaskar.ebay77@gmail.com> wrote:

> Hi
> In the following example given in flink:
> object ExampleCountWindowAverage extends App {
>   val env = StreamExecutionEnvironment.getExecutionEnvironment
>
>   env.fromCollection(List(
>     (1L, 3L),
>     (1L, 5L),
>     (1L, 7L),
>     (1L, 4L),
>     (1L, 2L)
>   )).keyBy(_._1)
>     .flatMap(new CountWindowAverage())
>     .print()
>   // the printed output will be (1,4) and (1,5)
>
>   env.execute("ExampleManagedState")
> }
>
> There is only 1 state because there is one key. In the CountWindowAverage
> method there is one state descriptor :  new ValueStateDescriptor[(Long,
> Long)]("average", createTypeInformation[(Long, Long)])
> Name given as "average". In order to implement this is generic way, shall
> i modify the  method:
>
> CountWindowAverage(keyName:String)  so that  new
> ValueStateDescriptor[(Long, Long)](keyName, createTypeInformation[(Long,
> Long)]) is created. But how to configure TTL for this? Inside this method?
> In the eample:
> https://ci.apache.org/projects/flink/flink-docs-stable/dev/stream/state/state.html#state-time-to-live-ttl
> ,   you have given a stand alone ValueStateDescriptor.  How can i use the
> TTL inside CountWindowAverage() per Key?
>
> Regards
> Bhaskar
>

Mime
View raw message