说明

用于合并数据，通过 DataStream 可以使用 union、connect 操作

union

union 算子可以合并多个 相同类型 的数据流，并生成同类型的数据流

数据按照 先进先出（First in first out） 合并

如下图，union对 白色 和 蓝色 两个数据流进行合并，生成一个数据流

例子

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

    DataStream<String> ds = env.fromElements("hadoop","yarn","flink");
    DataStream<String> ds2 = env.fromElements("lilei","hanmeimei","lucy");
    //合并
    DataStream<String> ds3=ds.union(ds2);
    ds3.print();
    env.execute();
}

connect

connect 只能连接2个数据流，数据类型 可 不一致，连接后转化为：ConnectedStreams

ConnectedStreams 会对 2 个流的数据应用不同的处理方法，且 2个流之间可共享状态

例子

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

    DataStream<String> ds = env.fromElements("hadoop","yarn","flink");
    DataStream<Integer> ds2 = env.fromElements(1,2,3);
    //先合并 2 个数据流
    ConnectedStreams<String, Integer> cs = ds.connect(ds2);
    //然后统一数据类型
    SingleOutputStreamOperator<String> mapResult = cs.map(new CoMapImpl());
    mapResult.print();
    env.execute();
}
/**
 CoMapFunction<String, Integer, String>泛型解释：
   第一个泛型 String：第一个集合的类型
   第二个泛型 Integer：第二个集合的类型
   第三个泛型 String：返回集合的类型，这里我们都转行成String

 */
static class CoMapImpl implements CoMapFunction<String, Integer, String>{

    /**
     * 处理第一个集合
     * @param s
     * @return
     * @throws Exception
     */
    @Override
    public String map1(String s) throws Exception {
        return s;
    }

    /**
     * 处理第二个集合
     * @param integer
     * @return
     * @throws Exception
     */
    @Override
    public String map2(Integer integer) throws Exception {
        return integer.toString();
    }
}