flink-user mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From Vishal Santoshi <vishal.santo...@gmail.com>
Subject Re: aggregate does not allow RichAggregateFunction ?
Date Wed, 06 Dec 2017 02:52:54 GMT
It seems that this has to do with session windows tbat are mergeable ? I
tried the RixhWindow function and that seems to suggest that one cannot use
state ? Any ideas folks...

On Dec 1, 2017 10:38 AM, "Vishal Santoshi" <vishal.santoshi@gmail.com>
wrote:

> I have a simple Aggregation with one caveat. For some reason I have to
> keep a large amount of state till the window is GCed. The state is within
> the Accumulator ( ACC ). I am hitting a memory bottleneck and would like to
> offload the state  to the states backend ( ROCKSDB), keeping the between
> checkpoint state in memory ( seems to be an obvious fix). I am not though
> allowed to have a RichAggregateFunction in the aggregate method of a
> windowed stream . That begs 2 questions
>
> 1. Why
> 2. Is there an alternative for stateful window aggregation where we manage
> the state. ?
>
> Thanks Vishal
>
>
> Here is the code ( generics but it works  )
>
> SingleOutputStreamOperator<OUT> retVal = input
>         .keyBy(keySelector)
>         .window(EventTimeSessionWindows.withGap(gap))
>         .aggregate(
>                 new AggregateFunction<IN, ACC, OUT>() {
>
>                     @Override
>                     public ACC createAccumulator() {
>                         ACC newInstance = (ACC) accumulator.clone();
>                         newInstance.resetLocal();
>                         return newInstance;
>                     }
>
>                     @Override
>                     public void add(IN value, ACC accumulator) {
>                         accumulator.add(value);
>
>                     }
>
>                     @Override
>                     public OUT getResult(ACC accumulator) {
>                         return accumulator.getLocalValue();
>                     }
>
>                     @Override
>                     public ACC merge(ACC a, ACC b) {
>                         a.merge(b);
>                         return a;
>                     }
>                 }, new WindowFunction<OUT, OUT, KEY, TimeWindow>() {
>                     @Override
>                     public void apply(KEY s, TimeWindow window, Iterable<OUT> input,
Collector<OUT> out) throws Exception {
>                             out.collect(input.iterator().next());
>                     }
>                 }, accType, aggregationResultType, aggregationResultType);
>
>

Mime
View raw message