1

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

  • 每个事件都单独处理（无窗口），
  • 每个事件都属于某个组，
  • 每个事件更新一个组状态，然后用于生成一些输出值。
public class IMapExample {
    public static void main(String[] args) {
        JetInstance jet = Jet.newJetInstance();
        IMap<Long, Double> groups = jet.getMap("groups");

        Pipeline p1 = Pipeline.create();
        p1.readFrom(TestSources.itemStream(10))
                .withoutTimestamps()
                .writeTo(Sinks.mapWithUpdating(groups,
                        event -> event.sequence() % 10, //simulate 10 groups
                        (oldState, event) -> event.sequence() + (oldState != null ? oldState : 0.0) //update group state with given event
                ));

        Pipeline p2 = Pipeline.create();
        p2.readFrom(Sources.mapJournal(groups, START_FROM_OLDEST))
                .withIngestionTimestamps()
                .map(x -> x.getKey() + " -> " + x.getValue()) //map group state to some output value
                .writeTo(Sinks.logger());

        jet.newJob(p2);
        jet.newJob(p1).join();
    }
}

鉴于上述示例，Hazelcast Jet 是否保留数据局部性？从某种意义上说，更新组状态的代码应该在给定组的状态所在的同一节点上调用。

后续问题：如果 StreamSource 被由相同组分区的 Kafka 替换，该分区是否会被保留并与管道的 Data Locality 相关联？

4

0 回答 0