MapState 在 RichCoFlatMapFunction 中始终为空

Question

我正在阅读 2 个流。一个带有记录，一个带有元数据。

我第一次希望我的应用程序通过扫描完整的表来构建元数据并将其保存到 Flink 的 MapState。表上的更新将通过元数据流捕获，并且 MapState 将相应更新。

从第二次开始，我想使用 MapState 而不是读取整个表格。

下面是我对这个功能的实现，但是我的 MapState 总是空的，我在这里做错了吗？

public class CustomCoFlatMap extends RichCoFlatMapFunction<Record, Metadata, Output> {

    private transient DataSource datasource;
    private transient MapState<String, Metadata> metadataState;

    @Inject
    public void setDataSource(DataSource datasource) {
        this.datasource = datasource;
    }

    @Override
    public void open(Configuration parameters) throws Exception {
        final RichFunctionComponent component = DaggerRichFunctionComponent.builder()
                .richFunctionModule(RichFunctionModule.builder()
                        .runtimeContext(getRuntimeContext())
                        .build())
                .build();
        component.inject(this);

        // read MapState from snapshot
        metadataState = getRuntimeContext().getMapState(new MapStateDescriptor<String, Cluster>("metadataState",
                TypeInformation.of(new TypeHint<String>(){}), TypeInformation.of(new TypeHint<Metadata>() {})));
    }

    @Override
    public void flatMap2(Metadata metadata, Collector<Output> collector) throws Exception {
        // this should happen only when application starts for first time
        // from next time, application will read from snapshot
        readMetadataForFirstTime();

        // update metadata in MapState
        this.metadataState.put(metadata.getId(), metadata);
    }

    @Override
    public void flatMap1(Record record, Collector<Output> collector) throws Exception {
        
        readMetadataForFirstTime();
        
        Metadata metadata = this.metadataState.get(record.getId());

        Output output = new Output(record.getId(), metadataState.getName(), metadataState.getVersion(), metadata.getType());
    
        collector.collect(output);
    }

    private void readMetadataForFirstTime() throws Exception {
        if(this.metadataState.iterator().hasNext()) {
            // metadataState from snapshot has data
            // not reading from table
            return;
        }

        // do this only once
        // read metadata from table and add it to MapState
        List<Metadata> metadataList = datasource.listAllMetadata();
        for(Metadata metadata: metadataList) {
            this.metadataState.put(metadata.getid(), metadata);
        }

    }
}

编辑：应用程序的其余部分

DataStream<Metadata> metadataKeyedStream =
                env.addSource(metadataStream)
                        .keyBy(Metadata::getId);

SingleOutputStreamOperator<Output> outputStream =
                env.addSource(recordStream)
                        .assignTimestampsAndWatermarks(new RecordTimeExtractor())
                        .keyBy(Record::getId)
                        .connect(metadataKeyedStream)
                        .flatMap(new CustomCoFlatMap());

Answer 1

MapState 是一种键分区状态——这意味着 Flink 为输入流中的每个不同键维护一个单独的 Map<String, Metadata>。 readMetadataForFirstTime 必须为您的RichCoFlatMapFunction 处理的流中的每个键读取并插入其数据，因为每个键都有一个单独的映射。

您可能希望以不同的方式处理此问题，具体取决于您要执行的操作。 例如，如果您只想为源流中的每个键存储一个值，那么您应该使用ValueState 而不是MapState。您可以将ValueState 视为分片键/值存储，其中有状态运算符的每个并行实例（例如，RichCoFlatMapFunction）将具有键空间切片的值。 MapState 适用于需要为每个键而不是单个对象存储整个哈希图的情况。

（如果我误判问题出在哪里，请分享更多上下文，说明应用程序的其余部分如何使用此RichCoFlatMapFunction。）