flink 流的合并

flink 流的合并操作

• union
  union只能合并类型相同的数据，合并的结果仍然是DataStream，结果操作与未合并之前一致。

  public static void main(String[] args) throws Exception {


        //流的合并操作  union 只能合并类型相同的流
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        DataStreamSource<String> ds1 = env.fromElements("night", "Jim", "Mary");

        DataStreamSource<String> ds2 = env.fromElements("四川", "北京", "上海");

        DataStream<String> union = ds1.union(ds2);

        union.print();
        env.execute();
    }

11> 北京
9> Mary
12> 上海
8> Jim
7> night
10> 四川

• connect
  connect可以连接不同类型的流，后续的处理api也有类似的不同，下列是一个tuple2与Long类型的流合并的结果，做了一个keyBy之后，在map的操作，map的实现接口是CoMapFunction

    public static void main(String[] args) throws Exception {

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        DataStreamSource<Tuple2<String, String>> ds1 = env.fromElements(Tuple2.of("四川", "成都"), Tuple2.of("北京", "朝阳"), Tuple2.of("广东", "深圳"),Tuple2.of("四川", "成都"));

        DataStreamSource<Long> ds2 = env.fromElements(1L, 2L, 3L,2L);


        ConnectedStreams<Tuple2<String, String>, Long> connect = ds1.connect(ds2);

        connect.keyBy(data -> data.f0,data -> data).map(new CoMapFunction<Tuple2<String, String>, Long, String>() {
//
            @Override
            public String map1(Tuple2<String, String> stringStringTuple2) throws Exception {
                return "this is tuple" + stringStringTuple2;
            }

            @Override
            public String map2(Long aLong) throws Exception {
                return "this is number" + aLong;
            }
        }).print();

        env.execute();

6> this is tuple(广东,深圳)
7> this is tuple(北京,朝阳)
15> this is number3
16> this is tuple(四川,成都)
11> this is number1
16> this is number2
16> this is tuple(四川,成都)
16> this is number2