Hazelcast Jet 是否允许从rollingAggregate 查询累加器值？

问题描述

下面的最小工作示例展示了一些（简化的）处理，其中：

• 每个事件都单独处理（无窗口），
• 每个事件都属于某个组，
• 每个事件更新一个组状态，然后用于生成一些输出值。

public class RollingExample {
    public static void main(String[] args) {
        JetInstance jet = Jet.newJetInstance();

        Pipeline p = Pipeline.create();
        p.readFrom(TestSources.itemStream(10))
                .withoutTimestamps()
                .groupingKey(event -> event.sequence() % 10) //simulates 10 groups
                .rollingAggregate(AggregateOperation
                        .withCreate(DoubleAccumulator::new) //group state simulated by a single double value
                        .andAccumulate((DoubleAccumulator acc,SimpleEvent event) -> acc.accumulate(event.sequence())) //update group state with given event
                        .andCombine(DoubleAccumulator::combine)
                        .andDeduct(DoubleAccumulator::deduct)
                        .andExportFinish(DoubleAccumulator::export))
                .map(x -> x.getKey() + " -> " + x.getValue()) //map group state to some output value
                .writeTo(Sinks.logger());

        jet.newJob(p).join();
    }
}

鉴于上面的例子，是否可以从外部查询单个组的状态，例如使用 Hazelcast 客户端？ “查询”是指以编程方式获取给定组的累加器的当前值。

解决方法

暂无找到可以解决该程序问题的有效方法，小编努力寻找整理中！

如果你已经找到好的解决方法，欢迎将解决方案带上本链接一起发送给小编。

小编邮箱:dio#foxmail.com (将#修改为@）