解决 Flink 窗口化 Keyed Join 无输出问题：深入理解与实践

joined_stream.print(); // 将结果打印到标准输出

import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.JoinFunction;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.connector.kafka.source.KafkaSource;
import org.apache.flink.connector.kafka.source.enumerator.initializer.OffsetsInitializer;
import org.apache.kafka.clients.consumer.ConsumerRecord;

import java.nio.charset.StandardCharsets;
import java.time.Duration;

public class FlinkKeyedJoinTutorial {

    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1); // 调试时可设置为1，方便观察

        // 假设 KafkaSource iotA 和 iotB 已经定义
        // 为了示例简洁，我们使用一个模拟的 KafkaSource 或直接从集合创建
        // 实际应用中，这里会配置你的 Kafka 连接信息和反序列化器

        // 模拟数据源 A
        DataStream<ConsumerRecord<String, String>> iotA_datastream = env.fromElements(
            new ConsumerRecord<>("iotA", 0, 0, 1678886400000L, null, 0L, 0, 0, "key1", "valueA1"), // 2023-03-15 00:00:00
            new ConsumerRecord<>("iotA", 0, 1, 1678886401000L, null, 0L, 0, 0, "key2", "valueA2"), // 2023-03-15 00:00:01
            new ConsumerRecord<>("iotA", 0, 2, 1678886403000L, null, 0L, 0, 0, "key1", "valueA3")  // 2023-03-15 00:00:03
        ).assignTimestampsAndWatermarks(
            WatermarkStrategy.<ConsumerRecord<String, String>>forMonotonousTimestamps()
                .withTimestampAssigner((record, timestamp) -> record.timestamp())
        );

        // 模拟数据源 B
        DataStream<ConsumerRecord<String, String>> iotB_datastream = env.fromElements(
            new ConsumerRecord<>("iotB", 0, 0, 1678886400500L, null, 0L, 0, 0, "key1", "valueB1"), // 2023-03-15 00:00:00.5
            new ConsumerRecord<>("iotB", 0, 1, 1678886402000L, null, 0L, 0, 0, "key2", "valueB2"), // 2023-03-15 00:00:02
            new ConsumerRecord<>("iotB", 0, 2, 1678886404000L, null, 0L, 0, 0, "key1", "valueB3")  // 2023-03-15 00:00:04
        ).assignTimestampsAndWatermarks(
            WatermarkStrategy.<ConsumerRecord<String, String>>forMonotonousTimestamps()
                .withTimestampAssigner((record, timestamp) -> record.timestamp())
        );

        // 假设原始问题中的 map 函数进行了某种值转换，这里简化
        DataStream<ConsumerRecord<String, String>> mapped_iotA = iotA_datastream
            .map(new MapFunction<ConsumerRecord<String, String>, ConsumerRecord<String, String>>() {
                @Override
                public ConsumerRecord<String, String> map(ConsumerRecord<String, String> record) throws Exception {
                    // 模拟一些转换，例如修改 value
                    String newValue = record.value() + "_processed";
                    return new ConsumerRecord<>(record.topic(), record.partition(), record.offset(),
                            record.timestamp(), record.timestampType(), record.checksum(),
                            record.serializedKeySize(), record.serializedValueSize(), record.key(), newValue);
                }
            })
            // 确保在转换后时间戳和水位线仍然有效，或者重新分配
            .assignTimestampsAndWatermarks(
                WatermarkStrategy.<ConsumerRecord<String, String>>forMonotonousTimestamps()
                    .withTimestampAssigner((record, timestamp) -> record.timestamp())
            );

        DataStream<ConsumerRecord<String, String>> mapped_iotB = iotB_datastream
            .map(new MapFunction<ConsumerRecord<String, String>, ConsumerRecord<String, String>>() {
                @Override
                public ConsumerRecord<String, String> map(ConsumerRecord<String, String> record) throws Exception {
                    String newValue = record.value() + "_transformed";
                    return new ConsumerRecord<>(record.topic(), record.partition(), record.offset(),
                            record.timestamp(), record.timestampType(), record.checksum(),
                            record.serializedKeySize(), record.serializedValueSize(), record.key(), newValue);
                }
            })
            .assignTimestampsAndWatermarks(
                WatermarkStrategy.<ConsumerRecord<String, String>>forMonotonousTimestamps()
                    .withTimestampAssigner((record, timestamp) -> record.timestamp())
            );

        // 执行 Keyed Join 操作
        DataStream<String> joined_stream = mapped_iotA.join(mapped_iotB)
                .where(new KeySelector<ConsumerRecord<String, String>, String>() {
                    @Override
                    public String getKey(ConsumerRecord<String, String> record) throws Exception {
                        return record.key();
                    }
                })
                .equalTo(new KeySelector<ConsumerRecord<String, String>, String>() {
                    @Override
                    public String getKey(ConsumerRecord<String, String> record) throws Exception {
                        return record.key();
                    }
                })
                .window(TumblingEventTimeWindows.of(Time.seconds(5))) // 5秒的滚动事件时间窗口
                .apply(new JoinFunction<ConsumerRecord<String, String>, ConsumerRecord<String, String>, String>() {
                    @Override
                    public String join(ConsumerRecord<String, String> record1, ConsumerRecord<String, String> record2) throws Exception {
                        // 此处代码将在匹配的事件发生时执行
                        System.out.println("Joined Result: Key=" + record1.key() +
                                ", ValueA=" + record1.value() +
                                ", ValueB=" + record2.value() +
                                ", TimestampA=" + record1.timestamp() +
                                ", TimestampB=" + record2.timestamp());
                        return "Joined: " + record1.key() + " -> " + record1.value() + " | " + record2.value();
                    }
                });

        // 关键步骤：为 joined_stream 添加一个 Sink
        // 如果没有这一行，JoinFunction 中的 System.out.println 将永远不会被触发
        joined_stream.print().name("Joined Output Sink");

        // 启动 Flink 作业
        env.execute("Flink Keyed Join Example with Sink");
    }
}