在Apache Flink中定制KafkaSource以读取键值对记录

import java.io.Serializable;

public class KafkaRecordInfo implements Serializable {
    public String key;
    public String value;
    public Long timestamp;
    public String topic;
    public int partition;
    public long offset;

    // Flink 需要一个无参构造函数来进行序列化和反序列化
    public KafkaRecordInfo() {}

    public KafkaRecordInfo(String key, String value, Long timestamp, String topic, int partition, long offset) {
        this.key = key;
        this.value = value;
        this.timestamp = timestamp;
        this.topic = topic;
        this.partition = partition;
        this.offset = offset;
    }

    @Override
    public String toString() {
        return "KafkaRecordInfo{" +
               "key='" + key + '\'' +
               ", value='" + value + '\'' +
               ", timestamp=" + timestamp +
               ", topic='" + topic + '\'' +
               ", partition=" + partition +
               ", offset=" + offset +
               '}';
    }
}

import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.connector.kafka.source.reader.deserializer.KafkaRecordDeserializationSchema;
import org.apache.flink.util.Collector;
import org.apache.kafka.clients.consumer.ConsumerRecord;

import java.nio.charset.StandardCharsets;

public class CustomKafkaRecordDeserializationSchema implements KafkaRecordDeserializationSchema<KafkaRecordInfo> {

    private static final long serialVersionUID = 1L; // 确保可序列化

    @Override
    public void deserialize(ConsumerRecord<byte[], byte[]> record, Collector<KafkaRecordInfo> out) throws Exception {
        // 提取键，如果键存在则转换为字符串，否则为null
        String key = (record.key() != null) ? new String(record.key(), StandardCharsets.UTF_8) : null;
        // 提取值，如果值存在则转换为字符串，否则为null
        String value = (record.value() != null) ? new String(record.value(), StandardCharsets.UTF_8) : null;
        // 提取时间戳
        Long timestamp = record.timestamp();
        // 提取主题
        String topic = record.topic();
        // 提取分区
        int partition = record.partition();
        // 提取偏移量
        long offset = record.offset();

        // 将提取的信息封装到 KafkaRecordInfo 对象中并发出
        out.collect(new KafkaRecordInfo(key, value, timestamp, topic, partition, offset));
    }

    @Override
    public TypeInformation<KafkaRecordInfo> getProducedType() {
        // 声明此反序列化器将生成 KafkaRecordInfo 类型的对象
        return TypeInformation.of(KafkaRecordInfo.class);
    }
}

import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.connector.kafka.source.KafkaSource;
import org.apache.flink.connector.kafka.source.enumerator.initializer.OffsetsInitializer;

public class FlinkKafkaKeyedRecordReader {

    public static void main(String[] args) throws Exception {
        // 设置 Flink 执行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1); // 示例设置为单并行度

        String bootstrapServers = "localhost:9092"; // Kafka 集群地址
        String topic = "test3"; // 你的 Kafka 主题
        String groupId = "flink-consumer-group"; // 消费者组ID

        // 构建 KafkaSource，使用自定义的反序列化器
        KafkaSource<KafkaRecordInfo> source = KafkaSource.<KafkaRecordInfo>builder()
                .setBootstrapServers(bootstrapServers)
                .setTopics(topic)
                .setGroupId(groupId)
                .setStartingOffsets(OffsetsInitializer.earliest()) // 从最早的偏移量开始消费
                .setDeserializer(new CustomKafkaRecordDeserializationSchema()) // 使用自定义反序列化器
                .build();

        // 从 KafkaSource 创建数据流
        DataStream<KafkaRecordInfo> stream = env.fromSource(source, WatermarkStrategy.noWatermarks(), "Kafka Keyed Source");

        // 对接收到的数据进行处理和打印
        stream.map(recordInfo -> "Received from Kafka: Key=" + recordInfo.key +
                                 ", Value=" + recordInfo.value +
                                 ", Timestamp=" + recordInfo.timestamp +
                                 ", Topic=" + recordInfo.topic +
                                 ", Partition=" + recordInfo.partition +
                                 ", Offset=" + recordInfo.offset)
              .print();

        // 执行 Flink 作业
        env.execute("Flink Kafka Keyed Record Reader");
    }
}