我有一个键控数据流，看起来像：

{
        summary:Integer
        uid:String
        key:String
        .....
    }

我需要在某个时间范围内聚合摘要值，一旦我获得了一个特定的数字，就会将摘要和所有影响摘要的uid刷新到数据库/日志文件中。
在第一次刷新之后，我想从内存中释放所有uid，并立即刷新每个新项。
所以我尝试了这个聚合函数。

public class AggFunc implements AggregateFunction<Item, Acc, Tuple2<Integer,List<String>>>{

    private static final long serialVersionUID = 1L;

    @Override
    public Acc createAccumulator() {
        return new Acc());
    }

    @Override
    public Acc add(Item value, Acc accumulator) {
        accumulator.inc(value.getSummary());
        accumulator.addUid(value.getUid);
        return accumulator;
    }

    @Override
    public Tuple2<Integer,List<String>> getResult(Acc accumulator) {
        List<String> newL = Lists.newArrayList(accumulator.getUids());
        accumulator.setUids(Lists.newArrayList());
        return Tuple2.of(accumulator.getSum(), newL);
    }

    @Override
    public Acc merge(Acc a, Acc b) {
        .....
    }

}

在aggregate process函数中，我将列表刷新为state，如果需要保存到数据库，我将清除state中的state和save标志以指示它。
但在我看来这是不公平的。我不确定这是否对我有效。
有没有更好的解决办法？