Flink practice -- multi stream merge

Flink actual combat – Multi stream merge

summary

​ In this paper, Flink Stream merge operation for . stay Flink in , The merging operators of streams are ：Union and Connect etc. . This paper mainly introduces the use of these two operators .

1.Union Use of operators

Return value ：DataStream->DataStream

function ： Merge two or more data streams , Create a new stream that contains all elements in all streams . Be careful : If you combine a data flow with itself , You will get each element in the result stream twice . in other words Union The operation will not remove the weight .

in addition , By union The data types of the two streams must be consistent .

explain ： adopt union Operator can merge two data streams , So when there are multiple data streams of the same type , You can do it. （ Merge ） To process as a stream . Thus, the flow processing is more flexible .

public class UionDemo1 {
    
    public static void main(String[] args) throws Exception {
    
        final StreamExecutionEnvironment senv =
                StreamExecutionEnvironment.getExecutionEnvironment();

        DataStream<Tuple2<String, Integer>> src1 = senv.fromElements(
                new Tuple2<>("shanghai", 15),
                new Tuple2<>("beijing", 25));

        DataStream<Tuple2<String, Integer>> src2 = senv.fromElements(
                new Tuple2<>("sichuan", 35),
                new Tuple2<>("chongqing", 45));

        DataStream<Tuple2<String, Integer>> src3 = senv.fromElements(
                new Tuple2<>("shenzheng", 55),
                new Tuple2<>("guanzhou", 65));
				
      	//  This stream cannot be combined with the above streams union, Due to the different types of stream data 
				// DataStream<Integer> src4 = senv.fromElements(2, 3);
      
        DataStream<Tuple2<String, Integer>> union = src1.union(src2, src3);

        union.filter(t->t.f1>30).print("union");
        senv.execute();
    }
}

The above code puts 3 Merge streams into one stream , Then process the merged stream （ Filter ）. Last printout . The output is as follows ：

union:5> (sichuan,35)
union:6> (chongqing,45)
union:6> (shenzheng,55)
union:7> (guanzhou,65)

2.connect Use of operators

Return value ：DataStream,DataStream → ConnectedStream

function ：“ Connect ” Two data streams and retain their respective types .connect Allows the sharing of state between the processing logic of two flows .

Connect Operator and Union The difference between operators ：

（1）Connect Operators can merge two different types of data streams , and Uion Only data streams of the same type can be merged .

（2）Connect The operator only supports the merging of two data streams ,union It can support the merging of multiple data streams .

（3） Two DataStream after connect Then it is transformed into ConnectedStreams,ConnectedStreams Different processing methods will be applied to the data of two streams , And the state can be shared between two streams .

Input and output ：ConnectedStream → DataStream
function ： It is similar to that on the connected data flow map and flatMap.

public class ConnectOpDemo1 {
    
    public static void main(String[] args) throws Exception {
    
        final StreamExecutionEnvironment senv =
                StreamExecutionEnvironment.getExecutionEnvironment();

        DataStream<Tuple2<String, Integer>> src1 = senv.fromElements(
                new Tuple2<>("shanghai", 15),
                new Tuple2<>("beijing", 25));
        
        DataStream<Integer> src4 = senv.fromElements(2, 3);

        ConnectedStreams<Tuple2<String, Integer>, Integer> connStream = src1.connect(src4);

        //  For different types of flows , Different treatment , And unified output into a new data type .
        //  here , I converted the data of both streams into String type , This facilitates subsequent processing .
        DataStream<String> res = connStream.flatMap(new CoFlatMapFunction<Tuple2<String, Integer>, Integer, String>() {
    
            @Override
            public void flatMap1(Tuple2<String, Integer> value, Collector<String> out) {
    
                out.collect(value.toString());
            }

            @Override
            public void flatMap2(Integer value, Collector<String> out) {
    
                String word = String.valueOf(value);
                out.collect(word);
            }
        });

        res.print();
        senv.execute();
    }

Summary

stay Flink Through Union and Connect Merge flows . From the above examples, we can see that the two operations are very convenient to use . But in use , You need to pay attention to the difference between them .