Flink多流转换

1、split和select

 

public static void main(String[] args) throws Exception {
    StreamExecutionEnvironment env=StreamExecutionEnvironment.getExecutionEnvironment();
    DataStreamSource<Long> streamSource = env.generateSequence(0, 10);
    SplitStream<Long> splitStream = streamSource.split(new OutputSelector<Long>() {
        List<String> outList = new ArrayList<>();
        @Override
        public Iterable<String> select(Long value) {
            if (value%2 == 0) {
                outList.add("even");
            } else {
                outList.add("odd");
            }
            return outList;
        }
    });
    splitStream.select("even").print("even=");
    splitStream.select("odd").print("odd=");
    splitStream.select("even","odd").print("even,odd=");
    env.execute(Test.class.getSimpleName());
}

2、connect和CoMap（特点：两条流的数据类型可以不一样）

connect：连接的两个流只是被放在同一个流中，内部依然保持各自的数据和形式，两个流相互独立。

public static void main(String[] args) throws Exception {
    StreamExecutionEnvironment env=StreamExecutionEnvironment.getExecutionEnvironment();
    List<Tuple2<Integer,String>> nameList=new ArrayList<>();
    nameList.add(new Tuple2<Integer,String>(1,"a"));
    nameList.add(new Tuple2<Integer,String>(2,"b"));
    List<Tuple2<Integer,Integer>> ageList=new ArrayList<>();
    ageList.add(new Tuple2<Integer,Integer>(1,10));
    ageList.add(new Tuple2<Integer,Integer>(2,20));
    DataStreamSource<Tuple2<Integer,String>> nameSource = env.fromCollection(nameList);
    DataStreamSource<Tuple2<Integer,Integer>> ageSource = env.fromCollection(ageList);
    ConnectedStreams<Tuple2<Integer, String>, Tuple2<Integer, Integer>> connect = nameSource.connect(ageSource);
    connect.map(new CoMapFunction<Tuple2<Integer, String>, Tuple2<Integer, Integer>, String>() {
        @Override
        public String map1(Tuple2<Integer, String> integerStringTuple2) throws Exception {
            return "name:"+integerStringTuple2.f0;
        }

        @Override
        public String map2(Tuple2<Integer, Integer> integerIntegerTuple2) throws Exception {
            return "age:"+integerIntegerTuple2.f0;
        }
    }).print();
    env.execute(Test.class.getSimpleName());
}

3、union（两个或者两个以上的dataStream的合并，但是多个流的数据类型必须一致）

public static void main(String[] args) throws Exception {
    StreamExecutionEnvironment env=StreamExecutionEnvironment.getExecutionEnvironment();
    DataStreamSource<Long> streamSource1 = env.generateSequence(0, 5);
    DataStreamSource<Long> streamSource2 = env.generateSequence(6, 10);
    DataStream<Long> union = streamSource1.union(streamSource2);
    union.print();
    env.execute(Test.class.getSimpleName());
}