Flink 中支持窗口上的多流合并, 需要保證的是輸入的 stream 要構(gòu)建在相同的 Window 上, 并使用相同類型的 Key 作為關(guān)聯(lián)條件.代碼如下所示, 先通過 join 方法將 inputStream1 數(shù)據(jù)集和 inputStream2 關(guān)聯(lián), 調(diào)用 where( ) 方法指定 inputStream1 的 key, 調(diào)用 equalTo( ) 方法指定 inputStream2 對(duì)應(yīng)關(guān)聯(lián)的 key. 通過 window( ) 方法指定 window Assigner, 最后再通過 apply( ) 方法傳入用戶自定義的 JoinFunction 或者 FlatJoinFunction 對(duì)輸入的數(shù)據(jù)元素做窗口計(jì)算.

inputStream1.join(inputStream2)
            // 指定inputStream1的關(guān)聯(lián)key
            .where(0)
            // 指定inputStream2的關(guān)聯(lián)key
            .equalTo(1)
            // 指定 window Assigner
            .window(TumblingEventTimeWindows.of(Time.seconds(10)))
            // 指定窗口計(jì)算函數(shù)
            .apply(<JoinFunction>)

下面就用 flink 官方倉庫中的join example來做演示, 完整代碼見倉庫 -> code link

樣例中有兩個(gè)流, 分別記錄的是員工的等級(jí)和員工的薪水, 流中數(shù)據(jù)的格式分別是 (name, grade) / (name, salary), 代碼實(shí)現(xiàn)的功能是合并兩個(gè)流, 轉(zhuǎn)變?yōu)?(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 兩個(gè)流 分別是 (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");

其中, 數(shù)據(jù)流的添加是通過一個(gè)Iterator 不停的添加進(jìn)去的, 具體的 join 邏輯通過 runWindowJoin( )方法, 以為為該方法的具體內(nèi)容

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);
                    }
                });
    }