Flink:如何通过key连接/分组2个流？

DataStream<Tuple2<String, Integer>> stream1 = ...; // 第一个流
DataStream<Tuple2<String, String>> stream2 = ...; // 第二个流

KeyedStream<Tuple2<String, Integer>, String> keyedStream1 = stream1.keyBy(data -> data.f0);
KeyedStream<Tuple2<String, String>, String> keyedStream2 = stream2.keyBy(data -> data.f0);

ConnectedStreams<Tuple2<String, Integer>, Tuple2<String, String>> connectedStreams = keyedStream1.connect(keyedStream2);

DataStream<Tuple2<Tuple2<String, Integer>, Tuple2<String, String>>> resultStream = connectedStreams
        .flatMap(new CoGroupFunction<Tuple2<String, Integer>, Tuple2<String, String>, Tuple2<Tuple2<String, Integer>, Tuple2<String, String>>>() {
            @Override
            public void coGroup(Iterable<Tuple2<String, Integer>> first, Iterable<Tuple2<String, String>> second, Collector<Tuple2<Tuple2<String, Integer>, Tuple2<String, String>>> out) {
                // 在这里可以对连接后的数据进行处理
                for (Tuple2<String, Integer> value1 : first) {
                    for (Tuple2<String, String> value2 : second) {
                        out.collect(Tuple2.of(value1, value2));
                    }
                }
            }
        });

resultStream.print();