flink 双流join

import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.api.common.functions.JoinFunction;
import org.apache.flink.api.java.tuple.Tuple2;

public class StreamJoinExample {
    public static void main(String[] args) throws Exception {
        final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        DataStream<Tuple2<String, Integer>> stream1 = env.fromElements(
            new Tuple2<>("a", 1),
            new Tuple2<>("b", 2),
            new Tuple2<>("c", 3)
        );

        DataStream<Tuple2<String, String>> stream2 = env.fromElements(
            new Tuple2<>("a", "Alice"),
            new Tuple2<>("b", "Bob"),
            new Tuple2<>("d", "David")
        );

        DataStream<String> joinedStream = stream1.join(stream2)
            .where(0) // 指定第一个流的键
            .equalTo(0) // 指定第二个流的键
            .window(TumblingProcessingTimeWindows.of(Time.seconds(5))) // 定义窗口
            .apply(new JoinFunction<Tuple2<String, Integer>, Tuple2<String, String>, String>() {
                @Override
                public String join(Tuple2<String, Integer> first, Tuple2<String, String> second) throws Exception {
                    return first.f0 + ": " + first.f1 + ", " + second.f1;
                }
            });

        joinedStream.print();

        env.execute("Flink Streaming Java API Skeleton");
    }
}