是否可以在 MapState 中拥有地图地图？

Question

private MapState<String, EventsHistory> eventsMap = null;
 
public void processElement2(Event event,
                            Context context,
                            Collector<JoinedEvent> collector) throws Exception {
    String name = event.getExperimentName();
    if (eventsMap.get(name) == null) {
         eventsMap.put(name, new EventsHistory());
    }
    eventsMap.get(name).put(event.getEventTime(), event);
}

class EventsHistory {

    private final Map<Long, Event> events = new HashMap<>();

    public Map<Long, Event> getEvents() {
        return events;
    }

    public void put(final Long eventTime, final Event event) {
        events.put(eventTime, event);
    }
}

我有上面的代码，想用Flink的MapState来维护一个map的map。 当我在本地测试时，我可以看到状态更新正常。但是当我在集群中运行它时，eventsMap 总是空的。

在MapState 中使用地图地图是否有效？有没有更好的方法来实现这一点？

作为替代方案，我尝试了以下版本，我自己进行分组。奇怪的是，这行得通。

private MapState<EventKey, Event> assignmentEventsMap = null;

public final class EventKey {

    private String name;
    private long eventTime;
}

    public void processElement2(Event event,
                                Context context,
                                Collector<JoinedEvent> collector) throws Exception {
        String name = event.getExperimentName();
        eventsMap
                .put(new EventKey(event.getName(), event.getEventTime()),
                        event);
    }

Answer 1

您分享的代码很难理解，但您可能误解了 MapState 是什么。 ValueState 提供了一个分片的键/值存储，分布在整个集群中。 MapState 为您提供了一个分片的键/值存储，其中值本身就是嵌套的 Map。

换句话说，MapState 始终是地图的地图。你最终试图创建一个地图的地图 - 这是一个太远的级别。

我假设您正在尝试构建这个结构，在其中您有效地拥有从实验名称到时间戳到事件的嵌套映射的映射：

name -> (time -> event)

假设您的事件流已经通过实验名称键入，那么您真正想要的是MapState<String, EventsHistory> eventsMap，而不是使用MapState<Long, Event> eventsMap，而不是

eventsMap.get(name).put(event.getEventTime(), event);

你应该这样做

eventsMap.put(event.getEventTime(), event);

有关如何使用这些机制的更多背景信息，请参阅 Flink 文档中的 the tutorial about ValueState 和 an example using MapState。