我想在整个数据集中做一个简单的最大值。我从卡夫卡的例子开始:https://github.com/hazelcast/hazelcast-jet-code-samples/blob/0.7-maintenance/kafka/src/main/java/avro/KafkaAvroSource.java
我刚刚将管道改为:
p.drawFrom(KafkaSources.<Integer, User>kafka(brokerProperties(), TOPIC))
.map(Map.Entry::getValue)
.rollingAggregate(minBy(comparingInt(user -> (Integer) user.get(2))))
.map(user -> (Integer) user.get(2))
.drainTo(Sinks.list("result"));
并转到:
IListJet<Integer> res = jet.getList("result");
SECONDS.sleep(10);
System.out.println(res.get(0));
SECONDS.sleep(15);
System.out.println(res.get(0));
cancel(job);
获得该主题中最大的人口年龄。然而,它不返回20并且似乎在不同的运行中返回不同的值。知道为什么吗?
您似乎正在使用rollingAggregate
,每次收到一些输入时都会生成一个新的输出项,但您检查的是它发出的第一个项目。您必须找到它发出的最新项目。实现它的一种方法是将结果推入IMap
接收器,每次使用相同的密钥:
p.drawFrom(KafkaSources.<Integer, User>kafka(brokerProperties(), TOPIC))
.withoutTimestamps()
.map(Map.Entry::getValue)
.rollingAggregate(minBy(comparingInt(user -> (Integer) user.get(2))))
.map(user -> entry("user", (Integer) user.get(2)))
.drainTo(Sinks.map("result"));
您可以使用获取最新结果
IMap<String, Integer> result = jet.getMap("result");
System.out.println(result.get("user");