Flink 中支持窗口上的多流合并, 需要保证的是输入的 stream 要构建在相同的 Window 上, 并使用相同类型的 Key 作为关联条件.代码如下所示, 先通过 join 方法将 inputStream1 数据集和 inputStream2 关联, 调用 where( ) 方法指定 inputStream1 的 key, 调用 equalTo( ) 方法指定 inputStream2 对应关联的 key. 通过 window( ) 方法指定 window Assigner, 最后再通过 apply( ) 方法传入用户自定义的 JoinFunction 或者 FlatJoinFunction 对输入的数据元素做窗口计算.

inputStream1.join(inputStream2)
            // 指定inputStream1的关联key
            .where(0)
            // 指定inputStream2的关联key
            .equalTo(1)
            // 指定 window Assigner
            .window(TumblingEventTimeWindows.of(Time.seconds(10)))
            // 指定窗口计算函数
            .apply(<JoinFunction>)

下面就用 flink 官方仓库中的join example来做演示, 完整代码见仓库 -> code link

样例中有两个流, 分别记录的是员工的等级和员工的薪水, 流中数据的格式分别是 (name, grade) / (name, salary), 代码实现的功能是合并两个流, 转变为 (name, grade, salary) 格式的流.

        final long windowSize = 200L;
        final long rate = 3L;

        System.out.println("Using windowSize=" + windowSize + ", data rate=" + rate);
        System.out.println("To customize example, use: WindowJoin [--windowSize <window-size-in-millis>] [--rate <elements-per-second>]");

        // 获取env, 配置为"ingestion time"
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setStreamTimeCharacteristic(TimeCharacteristic.IngestionTime);


        // 生成 grade 和 salary 两个流 分别是 (name, grade) / (name, salary)
        DataStream<Tuple2<String, Integer>> grades = WindowJoinSampleData.GradeSource.getSource(env, rate);
        DataStream<Tuple2<String, Integer>> salaries = WindowJoinSampleData.SalarySource.getSource(env, rate);

        DataStream<Tuple3<String, Integer, Integer>> joinedStream = runWindowJoin(grades, salaries, windowSize);

        joinedStream.print().setParallelism(1);

        env.execute("Windowed Join Example");

其中, 数据流的添加是通过一个Iterator 不停的添加进去的, 具体的 join 逻辑通过 runWindowJoin( )方法, 以为为该方法的具体内容

public static DataStream<Tuple3<String, Integer, Integer>> runWindowJoin(
            DataStream<Tuple2<String, Integer>> grades,
            DataStream<Tuple2<String, Integer>> salaries,
            long windowSize) {

        return grades.join(salaries)
                .where(new NameKeySelector())
                .equalTo(new NameKeySelector())

                .window(TumblingEventTimeWindows.of(Time.milliseconds(windowSize)))

                .apply(new JoinFunction<Tuple2<String, Integer>, Tuple2<String, Integer>, Tuple3<String, Integer, Integer>>() {

                    @Override
                    public Tuple3<String, Integer, Integer> join(
                            Tuple2<String, Integer> first,
                            Tuple2<String, Integer> second) {
                        return new Tuple3<String, Integer, Integer>(first.f0, first.f1, second.f1);
                    }
                });
    }