Flink Java 之数据流的 union 与 connect

Union
DataStream* → DataStream
将两个或多个数据流联合来创建一个包含所有流中数据的新流。
注意：如果一个数据流和自身进行联合，这个流中的每个数据将在合并后的流中出现两次。

Connect
DataStream,DataStream → ConnectedStream
“连接” 两个数据流并保留各自的类型。connect 允许在两个流的处理逻辑之间共享状态。

区别与联系
connect与union的区别在于：
connect只能连接两个数据流，union可以连接多个数据流。
connect所连接的两个数据流的数据类型可以不一致，union所连接的两个数据流的数据类型必须一致。
两个DataStream经过connect之后被转化为ConnectedStreams，ConnectedStreams会对两个流的数据应用不同的处理方法，且双流之间可以共享状态。

package com.daidai.transform.base;

import org.apache.flink.streaming.api.datastream.ConnectedStreams;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.co.CoMapFunction;

import java.util.Arrays;

public class UnionAndConnect {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        DataStreamSource source1 = env.fromCollection(Arrays.asList("hello", "how", "are", "you"));
        DataStreamSource source2 = env.fromCollection(Arrays.asList("java", "python", "scala"));
        DataStreamSource source3 = env.fromCollection(Arrays.asList(1, 2, 3, 4));

        source1.union(source2).print("union--string_string");

        ConnectedStreams connect = source1.connect(source3);
        connect.map(new CoMapFunction() {
            @Override
            public String map1(String value) throws Exception {
                return value;
            }

            @Override
            public String map2(Integer value) throws Exception {
                return value+"";
            }
        }).print("connect--string_int");


        env.execute();

    }
}