Flink Join 操作无输出：理解与解决 Flink 懒加载机制

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.KafkaDeserializationSchema;
import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.connector.kafka.source.KafkaSource;
import org.apache.flink.connector.kafka.source.enumerator.initializer.OffsetsInitializer;
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.kafka.clients.consumer.ConsumerRecord;

import java.nio.charset.StandardCharsets;

public class FlinkJoinOutputExample {

    // 假设 splitValue 方法存在，用于处理字符串
    private static String splitValue(String value, int index) {
        // 示例实现，根据实际需求调整
        String[] parts = value.split(",");
        if (parts.length > index) {
            return parts[index];
        }
        return value;
    }

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

        String IP = "localhost:9092"; // 替换为你的Kafka地址

        // Kafka Source for iotA
        KafkaSource<ConsumerRecord> iotA = KafkaSource.<ConsumerRecord>builder()
                .setBootstrapServers(IP)
                .setTopics("iotA")
                .setStartingOffsets(OffsetsInitializer.latest())
                .setDeserializer(KafkaRecordDeserializationSchema.of(new KafkaDeserializationSchema<ConsumerRecord>() {
                    @Override
                    public boolean isEndOfStream(ConsumerRecord record) { return false; }

                    @Override
                    public ConsumerRecord deserialize(ConsumerRecord<byte[], byte[]> record) throws Exception {
                        String key = new String(record.key(), StandardCharsets.UTF_8);
                        String value = new String(record.value(), StandardCharsets.UTF_8);
                        return new ConsumerRecord(
                                record.topic(), record.partition(), record.offset(), record.timestamp(),
                                record.timestampType(), record.checksum(), record.serializedKeySize(),
                                record.serializedValueSize(), key, value
                        );
                    }

                    @Override
                    public TypeInformation<ConsumerRecord> getProducedType() {
                        return TypeInformation.of(ConsumerRecord.class);
                    }
                }))
                .build();

        // Kafka Source for iotB (与iotA类似，省略具体实现)
        KafkaSource<ConsumerRecord> iotB = KafkaSource.<ConsumerRecord>builder()
                .setBootstrapServers(IP)
                .setTopics("iotB")
                .setStartingOffsets(OffsetsInitializer.latest())
                .setDeserializer(KafkaRecordDeserializationSchema.of(new KafkaDeserializationSchema<ConsumerRecord>() {
                    @Override
                    public boolean isEndOfStream(ConsumerRecord record) { return false; }

                    @Override
                                public ConsumerRecord deserialize(ConsumerRecord<byte[], byte[]> record) throws Exception {
                        String key = new String(record.key(), StandardCharsets.UTF_8);
                        String value = new String(record.value(), StandardCharsets.UTF_8);
                        return new ConsumerRecord(
                                record.topic(), record.partition(), record.offset(), record.timestamp(),
                                record.timestampType(), record.checksum(), record.serializedKeySize(),
                                record.serializedValueSize(), key, value
                        );
                    }

                    @Override
                    public TypeInformation<ConsumerRecord> getProducedType() {
                        return TypeInformation.of(ConsumerRecord.class);
                    }
                }))
                .build();

        // 从 Source 创建 DataStream 并分配时间戳和水位线
        DataStream<ConsumerRecord> iotA_datastream = env.fromSource(iotA,
                WatermarkStrategy.<ConsumerRecord>forMonotonousTimestamps()
                        .withTimestampAssigner((record, timestamp) -> record.timestamp()), "Kafka Source A");

        DataStream<ConsumerRecord> iotB_datastream = env.fromSource(iotB,
                WatermarkStrategy.<ConsumerRecord>forMonotonousTimestamps()
                        .withTimestampAssigner((record, timestamp) -> record.timestamp()), "Kafka Source B");

        // 对 DataStream 进行 Map 转换，并重新分配时间戳和水位线（如果需要更新时间戳逻辑）
        // 注意：此处如果时间戳逻辑不变，可以省略assignTimestampsAndWatermarks，直接使用上一步的。
        // 但如果map操作改变了事件时间相关的字段，则需要重新分配。
        DataStream<ConsumerRecord> mapped_iotA = iotA_datastream.map(new MapFunction<ConsumerRecord, ConsumerRecord>() {
            @Override
            public ConsumerRecord map(ConsumerRecord record) throws Exception {
                String new_value = splitValue((String) record.value(), 0);
                return new ConsumerRecord(record.topic(), record.partition(), record.offset(), record.timestamp(), record.timestampType(),
                        record.checksum(), record.serializedKeySize(), record.serializedValueSize(), record.key(), new_value);
            }
        }).assignTimestampsAndWatermarks(WatermarkStrategy.<ConsumerRecord>forMonotonousTimestamps()
                .withTimestampAssigner((record, timestamp) -> record.timestamp()));

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

        // 执行 Keyed Window Join 操作
        DataStream<String> joined_stream = mapped_iotA.join(mapped_iotB)
                .where(new KeySelector<ConsumerRecord, String>() {
                    @Override
                    public String getKey(ConsumerRecord record) throws Exception {
                        // System.out.println((String) record.key() + record.value()); // 调试信息
                        return (String) record.key();
                    }
                })
                .equalTo(new KeySelector<ConsumerRecord, String>() {
                    @Override
                    public String getKey(ConsumerRecord record) throws Exception {
                        // System.out.println((String) record.key() + record.value()); // 调试信息
                        return (String) record.key();
                    }
                })
                .window(TumblingEventTimeWindows.of(Time.seconds(5))) // 5秒翻滚事件时间窗口
                .apply(new JoinFunction<ConsumerRecord, ConsumerRecord, String>() {
                    @Override
                    public String join(ConsumerRecord record1, ConsumerRecord record2) throws Exception {
                        System.out.println("Joined: value1=" + record1.value() + ", value2=" + record2.value()); // 调试信息
                        return "Joined Result: A=" + record1.value() + ", B=" + record2.value();
                    }
                });

        // *** 关键步骤：添加 Sink 来消费 joined_stream 的结果 ***
        joined_stream.print("Joined Output"); // 将结果打印到标准输出，并带有标签

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