初始化状态导致此错误“java.lang.NullPointerException：Keyed state只能在'Keyed stream'上使用”

Question

我试图初始化CoProcessFunction中的“ListState”，但它不断抛出这个错误“java.lang.NullPointerException：Keyed state只能用于'Keyed stream'”“。

这是我用的密码

    dataStream1.keyBy(ele->ele.f1).connect(dataStream2).process(
        new CoProcessFunction<Tuple2<Long, String>, String, Object>() {
            ListState stream1ListState;
            
            @Override
            public void open(Configuration parameters) throws Exception {
                LOG.info("inside CoProcess Function Constructor");
                ListStateDescriptor<Tuple2<Long,String>> listStateDescriptor = new
                    ListStateDescriptor<Tuple2<Long,String>>("type1",TypeInformation.of(new TypeHint<Tuple2<Long,String>>(){}));
                stream1ListState = getRuntimeContext().getListState(listStateDescriptor);
            }
            
            @Override
            public void processElement1(Tuple2<Long, String> longStringTuple2, Context context,
                Collector<Object> collector) throws Exception {
                LOG.info("inside stream1 processor");
                LOG.info(longStringTuple2.toString());
                collector.collect(longStringTuple2);
            }
            
            @Override
            public void processElement2(String s, Context context, Collector<Object> collector)
                throws Exception {
                LOG.info("inside stream2 processor");
                LOG.info(s);
            }
        }).print();

引发此错误的行是

stream1ListState = getRuntimeContext().getListState(listStateDescriptor);

错误日志跟踪的其余部分如下所示

Caused by: java.lang.NullPointerException: Keyed state can only be used on a 'keyed stream', i.e., after a 'keyBy()' operation.
    at org.apache.flink.util.Preconditions.checkNotNull(Preconditions.java:76)
    at org.apache.flink.streaming.api.operators.StreamingRuntimeContext.checkPreconditionsAndGetKeyedStateStore(StreamingRuntimeContext.java:232)
    at org.apache.flink.streaming.api.operators.StreamingRuntimeContext.getListState(StreamingRuntimeContext.java:202)

Answer 1

该错误意味着您只能在KeyedCoProcessFunction内部使用ListState、ValueState等，而这些内容只能在两个流都被键控时使用。因此，如果第二个流是正常流，并且可以通过某个键进行键控，那么您可以这样做，否则您可能需要参考这里所述的广播状态模式（Broadcast state pattern）。