Flink：我如何使用键控状态？

Question

我有一个带有一些键的流，我想为每个键存储一些状态。我的流看起来如下所示：

inputStream
    .keyBy(new MyKeySelector())
    .process(new MyKeyedProcessFunction());

在KeyedProcessFunction中,我有一个状态变量：

public class MyKeyedProcessFunction extends KeyedProcessFunction<...> {

    private MapState<String, ...> state;

    @Override
    public void open(Configuration config) {

        MapStateDescriptor<String, ...> descriptor = new MapStateDescriptor<>(
            "keyed",
            BasicTypeInfo.STRING_TYPE_INFO,
            TypeInformation.of(new TypeHint<...>() {
            }));

        state = getRuntimeContext().getMapState(descriptor);
    }

    @Override
    public void processElement(... event, Context context, Collector<...> out) throws Exception {

        String key = context.getCurrentKey();

        ... keyedState;
        if (state.contains(key)) {
            keyedState = state.get(key);
        } else {
            keyedState = new ...();
        }
        ...
    }

...
}

我对此还这么陌生，我做错了什么？

Answer 1

这在作为Flink文档一部分的教程中有很好的介绍，在Ververica共享的免费Flink培训材料中也有更深入的介绍。这两个课程都利用了https://github.com/apache/flink-training中的练习，这将使所有这些更加清晰。

如果您的目标是每个键存储一个对象，那么您所需要的就是valueState ，它将创建一个分布在集群中的分片散列映射，为每个不同的键存储一个t类型的对象。mapstate用于需要为每个键存储一个hashmap时--例如，如果您希望为每个用户都有一个开放式属性哈希，给定一个由userid键控的流。

KeyedProcessFunction的实例在属于分配给运算符实例的密钥组的所有密钥上复用。在KeyedProcessFunction的Open方法中，上下文中没有键；当您在那里实例化valueState对象时，您将取回内存中或本地磁盘上的hashmap句柄，这取决于您使用的后端状态。然后，当您在processElement()方法中调用state.value()或state.update(...)时，上下文中有一个特定事件，当前事件的键被隐式地用于读取或写入状态HashMap中的适当条目。

此接口的设计目的是防止您试图操作无法在本地实例中访问的键的状态（由于该状态是在集群中分片的，因此不能保证该实例中除了当前事件的键之外的任何其他键的状态可用）。但不可否认的是，关键字在API中却不可见，却发挥着如此重要的作用，这是令人困惑的。