如何实现支持快速插入、查找和迭代嵌套映射的Flink键控状态的映射？

import org.apache.flink.api.common.functions.RichFlatMapFunction;
import org.apache.flink.api.common.state.MapState;
import org.apache.flink.api.common.state.MapStateDescriptor;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.util.Collector;

public class NestedMapStateExample extends RichFlatMapFunction<Tuple2<String, String>, String> {
    private transient MapState<String, MapState<String, String>> nestedMapState;
    private transient ValueState<Integer> countState;

    @Override
    public void open(Configuration parameters) throws Exception {
        MapStateDescriptor<String, MapState<String, String>> nestedMapStateDescriptor =
                new MapStateDescriptor<>("nestedMapState", String.class, new MapStateDescriptor<>("innerMapState", String.class, String.class));
        nestedMapState = getRuntimeContext().getMapState(nestedMapStateDescriptor);

        ValueStateDescriptor<Integer> countStateDescriptor =
                new ValueStateDescriptor<>("countState", Integer.class);
        countState = getRuntimeContext().getState(countStateDescriptor);
    }

    @Override
    public void flatMap(Tuple2<String, String> input, Collector<String> collector) throws Exception {
        String key = input.f0;
        String value = input.f1;

        // 获取或创建外层MapState
        MapState<String, String> innerMapState = nestedMapState.get(key);
        if (innerMapState == null) {
            innerMapState = getRuntimeContext().getMapState(new MapStateDescriptor<>(key, String.class, String.class));
            nestedMapState.put(key, innerMapState);
        }

        // 向内层MapState插入键值对
        innerMapState.put(value, value);

        // 获取或创建计数状态
        Integer count = countState.value();
        if (count == null) {
            count = 0;
        }

        // 更新计数状态
        count++;
        countState.update(count);

        collector.collect("Processed " + count + " records");
    }
}