不能每次都从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

广播状态元素和正常元素到达运算符的顺序不能保证始终以相同的顺序发生。这取决于生成元素的上游运算符。这就是为什么您有时会看到完整的输出，有时只看到没有广播状态元素的输出(所有正常元素都到达广播状态元素之前(。

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