Flink 通过Window CoGroup 实现full outer join,join,left join,right join

stream1.coGroup(stream2)
.where(<KeySelector>)
.equalTo(<KeySelector>)
.window(TumblingEventTimeWindows.of(Time.hours(1)))
.apply(<CoGroupFunction>)

• 具体代码实现

import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.CoGroupFunction;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.util.Collector;

/**
 * author :
 * date   : 2022/5/31 0031
 * Desc   : join, left join, right join
 * modified by :
 * version : 1.0
 */
public class CoGroupExample {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env =
                StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        SingleOutputStreamOperator<Tuple2<String, Long>> stream1 = env
                .fromElements(
                        Tuple2.of("a", 1000L),
                        Tuple2.of("b", 1000L),
                        Tuple2.of("a", 2000L),
                        Tuple2.of("b", 2000L),
                        Tuple2.of("d", 2000L),
                        Tuple2.of("d", 2000L)
                ).assignTimestampsAndWatermarks(
                        WatermarkStrategy.<Tuple2<String, Long>>forMonotonousTimestamps()
                                .withTimestampAssigner(new SerializableTimestampAssigner<Tuple2<String, Long>>() {
                                    @Override
                                    public long extractTimestamp(Tuple2<String, Long> element, long recordTimestamp) {
                                        return element.f1;
                                    }
                                })

                );


        DataStream<Tuple2<String, Long>> stream2 = env
                .fromElements(
                        Tuple2.of("a", 3000L),
                        Tuple2.of("b", 3000L),
                        Tuple2.of("a", 4000L),
                        Tuple2.of("b", 4000L),
                        Tuple2.of("c", 4000L),
                        Tuple2.of("c", 4000L)
                )
                .assignTimestampsAndWatermarks(
                        WatermarkStrategy
                                .<Tuple2<String, Long>>forMonotonousTimestamps()
                                .withTimestampAssigner(
                                        new SerializableTimestampAssigner<Tuple2<String, Long>>() {
                                            @Override
                                            public long extractTimestamp(Tuple2<String, Long> element, long recordTimestamp) {
                                                return element.f1;
                                            }
                                        }
                                )
                );

        // full outer join
        stream1.coGroup(stream2)
                .where(data1 -> data1.f0)
                .equalTo(data2 -> data2.f0)
                .window(TumblingEventTimeWindows.of(Time.seconds(5)))
                .apply(new CoGroupFunction<Tuple2<String, Long>, Tuple2<String, Long>, String>() {
                    @Override
                    public void coGroup(Iterable<Tuple2<String, Long>> first, Iterable<Tuple2<String, Long>> second, Collector<String> out) throws Exception {
                        //join
                            out.collect("full outer join: " + first + " => " + second);
                    }
                })
                .print("full outer join");
        System.out.println("----------------------------------");


        // join
        stream1.coGroup(stream2)
                .where(data1 -> data1.f0)
                .equalTo(data2 -> data2.f0)
                .window(TumblingEventTimeWindows.of(Time.seconds(5)))
                .apply(new CoGroupFunction<Tuple2<String, Long>, Tuple2<String, Long>, String>() {
                    @Override
                    public void coGroup(Iterable<Tuple2<String, Long>> first, Iterable<Tuple2<String, Long>> second, Collector<String> out) throws Exception {
                        //join
                        if (first.iterator().hasNext() && second.iterator().hasNext()) {
                            out.collect("join: " + first + " => " + second);
                        }
                    }
                })
                .print("join");
        System.out.println("----------------------------------");

        // left join
        stream1.coGroup(stream2)
                .where(data1 -> data1.f0)
                .equalTo(data2 -> data2.f0)
                .window(TumblingEventTimeWindows.of(Time.seconds(5)))
                .apply(new CoGroupFunction<Tuple2<String, Long>, Tuple2<String, Long>, String>() {
                    @Override
                    public void coGroup(Iterable<Tuple2<String, Long>> first, Iterable<Tuple2<String, Long>> second, Collector<String> out) throws Exception {
                        //left join
                        if (first.iterator().hasNext()) {
                            out.collect("left join: " +first + " => " + second);
                        }
                    }
                })
                .print("left join");

        System.out.println("----------------------------------");

        // right join
        stream1.coGroup(stream2)
                .where(data1 -> data1.f0)
                .equalTo(data2 -> data2.f0)
                .window(TumblingEventTimeWindows.of(Time.seconds(5)))
                .apply(new CoGroupFunction<Tuple2<String, Long>, Tuple2<String, Long>, String>() {
                    @Override
                    public void coGroup(Iterable<Tuple2<String, Long>> first, Iterable<Tuple2<String, Long>> second, Collector<String> out) throws Exception {
                        //right join
                        if (second.iterator().hasNext()) {
                            out.collect("right join: " +first + " => " + second);
                        }
                    }
                })
                .print("right join");

        System.out.println("----------------------------------");


        env.execute();
    }
}