HazelcastJet滚动-聚合，删除以前的数据并添加新的数据

Question

Balance 1 {id:1, amount:100} // aggregated result 100
Balance 2 {id:2, amount:200} // 300
Balance 3 {id:1, amount:400} // 600 after removing 100 and adding 400

我可以在每次添加的地方实现简单的使用。但我无法实现需要减去现有价值而增加新价值的聚合。

rollingAggregation(AggregatorOperations.summingDouble(<login to add remove>))
    .drainTo(Sinks.logger()).

1. 余额1、2、3是邮件的序列
2. 注释显示Jet执行的每个消息的聚合值。
3. 我的目标是添加新的金额（如果第一次来了id）和减去金额，如果一个更新的余额来了I。E.Id和之前一样。

Answer 1

您可以尝试一个自定义聚合操作，它将发出以前和当前看到的值，如下所示：

public static <T> AggregateOperation1<T, ?, Tuple2<T, T>> previousAndCurrent() {
    return AggregateOperation
            .withCreate(() -> new Object[2])
            .<T>andAccumulate((acc, current) -> {
                acc[0] = acc[1];
                acc[1] = current;
            })
            .andExportFinish((acc) -> tuple2((T) acc[0], (T) acc[1]));
}

输出应该是（previous,current)形式的元组。然后可以再次对输出应用滚动集料。为了简化输入问题，我有一对(id,amount)对。

Pipeline p = Pipeline.create();
p.drawFrom(Sources.<Integer, Long>mapJournal("map", START_FROM_OLDEST)) // (id, amount)
        .groupingKey(Entry::getKey)
        .rollingAggregate(previousAndCurrent(), (key, val) -> val)
        .rollingAggregate(AggregateOperations.summingLong(e -> {
            long prevValue = e.f0() == null ? 0 : e.f0().getValue();
            long newValue = e.f1().getValue();
            return newValue - prevValue;
        }))
        .drainTo(Sinks.logger());

JetConfig config = new JetConfig();
config.getHazelcastConfig().addEventJournalConfig(new EventJournalConfig().setMapName("map"));
JetInstance jet = Jet.newJetInstance(config);

IMapJet<Object, Object> map = jet.getMap("map");

map.put(0, 1L);
map.put(0, 2L);
map.put(1, 10L);
map.put(1, 40L);

jet.newJob(p).join();

这应该作为输出产生:1,2,12,42。