flink使用07-通过join合并流的操作

发表于 | 分类于

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

1
2
3
4
5
6
7
8
9
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) 格式的流.

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
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( )方法, 以为为该方法的具体内容

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
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);
                    }
                });
    }
0%