我有下面的flink工作，在那里我尝试使用keyed stream stateful函数（mapstate）和后端类型rockdb，

environment
.addSource(consumer).name("MyKafkaSource").uid("kafka-id")
.flatMap(pojoMapper).name("MyMapFunction").uid("map-id")
.keyBy(new MyKeyExtractor())
.map(new MyRichMapFunction()).name("MyRichMapFunction").uid("rich-map-id")
.addSink(sink).name("MyFileSink").uid("sink-id")

MyricMapFunction是一个有状态函数，它扩展了richmapfunction，richmapfunction有以下代码，

public static class MyRichMapFunction extends RichMapFunction<MyEvent, MyEvent> {
    private transient MapState<String, Boolean> cache;
    @Override
    public void open(Configuration config) {
        MapStateDescriptor<String, Boolean> descriptor =
                new MapStateDescriptor("seen-values", TypeInformation.of(new TypeHint<String>() {}), TypeInformation.of(new TypeHint<Boolean>() {}));
        cache = getRuntimeContext().getMapState(descriptor);
    }
    @Override
    public MyEvent map(MyEvent value) throws Exception {
        if (cache.contains(value.getEventId())) {
            value.setIsSeenAlready(Boolean.TRUE);
            return value;
        }
        value.setIsSeenAlready(Boolean.FALSE);
        cache.put(value.getEventId(), Boolean.TRUE)
        return value;
    }
}

在将来，我想重新缩放并行度（从2到4），所以我的问题是，如何实现可重新缩放的键控状态，以便在更改并行度之后，我可以将相应的缓存键控数据获取到其相应的任务槽。我试图探索这个，在这里我发现了一个文档。据此，可以通过listscheckpointed接口实现可重扩展的操作员状态，listscheckpointed接口提供了snapshotstate/restorestate方法。但不确定如何实现可伸缩的键控状态（myrichmapfunction）？我是否需要为myrichmapfunction类实现listcheckpointed接口？如果是，我如何根据restorestate方法上的新并行密钥哈希重新分配缓存（我的mapstate将在启用ttl的情况下保存大量密钥，假设它在任何时间点都最多保存10亿个密钥）？有人能帮我一下吗？或者你能给我举个很好的例子吗。