同时在两个DataStream之前，在flink中找到一个DataStream包含来自其他DataStream的值？

Question

假设我有两个 DataStream

DataStream<String> A contains values {id1_id2 , id3_id4, id99_id0, id15_id3,id11_id5....}

DataStream<String> B contains values {id2, id3,id5...}

是否可以在DataStream A上进行一些处理，以便它将在另一个中输出值

DataStream<String> C ={id1, id3, id15, id11}

所以B中存在的所有值都将与a相交。我已经尝试使用processElement()和RichCoFlatMapFunction，但它不起作用。

public class MatchAggregator
        extends RichCoFlatMapFunction<String, String, Tuple1<String>> {

    private ValueState<String> doubleState;
    private ValueState<String> singleState;

    @Override
    public void open(Configuration config) {

        doubleState = getRuntimeContext().getState(new ValueStateDescriptor<>("doubleEvents",String.class));
        singleState = getRuntimeContext().getState(new ValueStateDescriptor<>("singleEvents",String.class));
    }
    
    @Override
    public void flatMap1(String s, Collector<Tuple1<String>> collector) throws Exception {
        String single = singleState.value();
       //this is outputting null.
        System.out.println(single);
      //s is also null
        if(single.contains(s)){
            String replaceNumber = single.replace(s,"");
            String replaceEmp = replaceNumber.replace("_","");
            single.clear();
            collector.collect(Tuple1.of(replaceEmp));
        }else {
            personContactState.update(s);
        }
    }

    @Override
    public void flatMap2(String s, Collector<Tuple1<String>> collector) throws Exception {

        
    }
}

我使用了两种数据流，比如:

DataStream<Tuple1<String>> match = A.connect(B).flatMap(new MatchAggregator());

match.print();




Answer 1

RichCoFlatMapFunction的确切行为将取决于您对两个连接流的键控方式。

String single=singleState。value（）。在您共享的代码中，update从不在singleState，因此singleState。value（）将始终为空。