每次都不能从processElement函数中使用BroadCastState

Question

我想第一次使用BroadCastState。我按照文档中的一个小例子测试了它。我使用了KeyedBroadcastProcessFunction并从processBroadcastElement函数更新了Map状态，但是当我试图从processElement函数获取状态来收集它时。有时它会输出要求，有时则不输出任何要求。这背后的原因是什么？

这是使用的代码。

DataStream<Tuple4<String,String,Integer,Integer>> similarityTuples = inputStream
                .keyBy(1)
                .connect(usersBroadCasted)
                .process(new KeyedBroadcastProcessFunction<String, Tuple3<String,String,Float>, String, Tuple4<String,String,Integer,Integer>>() {

                    MapStateDescriptor<Integer, String> usersBroadcastState =
                            new MapStateDescriptor<>(
                                    //"patterns", BasicTypeInfo.VOID_TYPE_INFO, BasicTypeInfo.STRING_TYPE_INFO);
                                    "patterns", BasicTypeInfo.INT_TYPE_INFO, BasicTypeInfo.STRING_TYPE_INFO);

                    ListState<String> usersLikedItem;

                    @Override
                    public void processElement(Tuple3<String, String, Float> input, ReadOnlyContext readOnlyContext, Collector<Tuple4<String, String, Integer, Integer>> out) throws Exception {

                        for(String user : usersLikedItem.get()){
                          out.collect(Tuple4.of(user,input.f0,1,0));
                        }
                        usersLikedItem.add(input.f0);

                        for (Map.Entry<Integer, String> entry : readOnlyContext.getBroadcastState(usersBroadcastState).immutableEntries()){                **out.collect(Tuple4.of(input.f0,entry.getValue(),0,10000));**
                        }
                    }

                    @Override
                    public void processBroadcastElement(String s, Context context, Collector<Tuple4<String, String, Integer, Integer>> collector) throws Exception {

                        context.getBroadcastState(usersBroadcastState).put(0,s);

                    }

我期待这个输出，它有时会输出我所期望的，而不会改变代码中的任何内容（必需）

(10,40,0,10000)
(10,20,1,0)
(20,40,0,10000)
(10,30,1,0)
(20,30,1,0)
(30,40,0,10000)
(40,40,0,10000)

但它有时输出以下内容

(10,20,1,0)
(10,30,1,0)
(20,30,1,0)

Answer 1

广播状态元素和普通元素到达运营商的顺序不能保证总是以相同的顺序发生。它取决于生成元素的上游运算符。这就是为什么你有时会看到完整输出，有时只看到没有广播状态元素的输出（所有普通元素都在广播状态元素之前到达）的原因。

如果你想保证你已经看到所有元素到达某个点，你需要等待水印（并且还生成它们），并且只有当你看到相应的水印表示没有更多元素时才需要处理这些元素。低于水印的时间戳将到达。