在 Apache Flink 中高效读取带键（Keyed）的 Kafka 记录

import java.io.Serializable;

public class KeyedKafkaRecord implements Serializable {
    private String key;
    private String value;
    private long timestamp;
    // 可根据需要添加其他元数据，例如 topic, partition, offset

    public KeyedKafkaRecord() {} // Flink POJO 要求无参构造函数

    public KeyedKafkaRecord(String key, String value, long timestamp) {
        this.key = key;
        this.value = value;
        this.timestamp = timestamp;
    }

    public String getKey() { return key; }
    public void setKey(String key) { this.key = key; }

    public String getValue() { return value; }
    public void setValue(String value) { this.value = value; }

    public long getTimestamp() { return timestamp; }
    public void setTimestamp(long timestamp) { this.timestamp = timestamp; }

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

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 org.apache.kafka.common.serialization.StringDeserializer;

import java.io.IOException;

public class CustomKeyedKafkaDeserializationSchema implements KafkaRecordDeserializationSchema<KeyedKafkaRecord> {

    private transient StringDeserializer keyDeserializer;
    private transient StringDeserializer valueDeserializer;

    @Override
    public void open(KafkaRecordDeserializationSchema.InitializationContext context) throws Exception {
        // 在反序列化器初始化时创建 Kafka Deserializer 实例
        keyDeserializer = new StringDeserializer();
        valueDeserializer = new StringDeserializer();
        // 如果需要配置，可以在这里进行，例如 keyDeserializer.configure(configs, true);
    }

    @Override
    public void deserialize(ConsumerRecord<byte[], byte[]> record, Collector<KeyedKafkaRecord> out) throws IOException {
        // 使用 Kafka StringDeserializer 反序列化键和值
        String key = keyDeserializer.deserialize(record.topic(), record.headers(), record.key());
        String value = valueDeserializer.deserialize(record.topic(), record.headers(), record.value());
        long timestamp = record.timestamp(); // 获取记录的时间戳

        // 将反序列化后的数据封装到自定义的 DTO 中
        out.collect(new KeyedKafkaRecord(key, value, timestamp));
    }

    @Override
    public TypeInformation<KeyedKafkaRecord> getProducedTypeInfo() {
        // 返回反序列化器产生的数据类型信息
        return TypeInformation.of(KeyedKafkaRecord.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 FlinkKeyedKafkaConsumerJob {
    public static void main(String[] args) throws Exception {
        // 获取 Flink 执行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

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

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

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

        // 对接收到的数据进行处理并打印
        stream.map(record -> "Received Key: " + record.getKey() +
                             ", Value: " + record.getValue() +
                             ", Timestamp: " + record.getTimestamp())
              .print();

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