自定义Flink KafkaSource读取键值对数据

本文详细阐述了如何在 Apache Flink 中使用 `KafkaSource` 读取包含键（Key）的 Kafka 记录。通过实现自定义的 `KafkaRecordDeserializationSchema`，用户可以完全访问 `ConsumerRecord` 对象，从而灵活地提取并处理记录的键、值、时间戳、分区、偏移量及其他元数据，克服了默认 `valueOnly` 反序列化器仅能获取记录值的局限性。

理解默认 valueOnly 反序列化器的局限性

在使用 Apache Flink 的 KafkaSource 从 Kafka 读取数据时，常见的做法是利用内置的 KafkaRecordDeserializationSchema.valueOnly(StringDeserializer.class) 来反序列化记录。这种方法简单便捷，但它仅将 Kafka 记录的值部分反序列化为指定类型（例如 String），而忽略了记录的键、时间戳、分区、偏移量以及任何附加的头部信息。

对于生产者以键值对形式发送的 Kafka 记录，例如通过 kafka-console-producer.sh --property "parse.key=true" --property "key.separator=:" 生产的数据，如果仅使用 valueOnly 反序列化器，我们将无法在 Flink 应用程序中获取到这些关键的键信息，也无法访问记录的时间戳等其他元数据。

解决方案：实现自定义 KafkaRecordDeserializationSchema

要解决上述局限性，核心在于实现一个自定义的 KafkaRecordDeserializationSchema。这个接口允许我们完全控制 Kafka ConsumerRecord 的反序列化过程。在 deserialize 方法中，我们可以直接访问到原始的 ConsumerRecord 对象，进而提取其键、值、时间戳等所有可用信息，并将其封装成 Flink 应用程序所需的任何自定义类型。

1. 定义一个用于承载记录信息的POJO

首先，我们定义一个简单的 Java POJO（Plain Old Java Object）来封装从 Kafka ConsumerRecord 中提取出的所有相关信息。这个POJO将作为自定义反序列化器的输出类型。

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 +
               '}';
    }
}

2. 实现自定义 KafkaRecordDeserializationSchema

接下来，创建 KafkaRecordDeserializationSchema 的实现。在这个实现中，我们将重写 deserialize 方法来解析 ConsumerRecord，并重写 getProducedType 方法来声明输出类型。

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 {

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

    @Override
    public void deserialize(ConsumerRecord record, Collector 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 getProducedType() {
        // 声明此反序列化器将生成 KafkaRecordInfo 类型的对象
        return TypeInformation.of(KafkaRecordInfo.class);
    }
}

3. 将自定义反序列化器集成到 KafkaSource

最后，将这个自定义的 CustomKafkaRecordDeserializationSchema 实例传递给 KafkaSource.builder().setDeserializer() 方法。

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

        // 从 KafkaSource 创建数据流
        DataStream 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");
    }
}

运行示例

1. 启动 Kafka 和 Flink： 确保 Kafka 集群正在运行，并且 Flink 环境已准备就绪。
2. 生产带键的 Kafka 消息： 使用 Kafka 控制台生产者发送带键的消息到 test3 主题：

   bin/kafka-console-producer.sh --topic test3 --property "parse.key=true" --property "key.separator=:" --bootstrap-server localhost:9092

   输入消息，例如：

   key1:valueA
   key2:valueB
   anotherKey:anotherValue

3. 运行 Flink 应用程序： 编译并运行上述 FlinkKafkaKeyedRecordReader Flink 应用程序。你将在 Flink 任务管理器的日志中看到类似以下的输出：

   Received from Kafka: Key=key1, Value=valueA, Timestamp=1678886400000, Topic=test3, Partition=0, Offset=0
   Received from Kafka: Key=key2, Value=valueB, Timestamp=1678886401000, Topic=test3, Partition=0, Offset=1
   Received from Kafka: Key=anotherKey, Value=anotherValue, Timestamp=1678886402000, Topic=test3, Partition=0, Offset=2

   这表明 Flink 成功地读取并解析了 Kafka 记录的键、值和时间戳等信息。

注意事项与扩展

• 错误处理： 在 deserialize 方法中，如果 byte[] 无法正确转换为 String（例如，编码不一致），可能会抛出异常。在实际生产环境中，应加入健壮的错误处理逻辑，例如使用 try-catch 块，将解析失败的记录发送到死信队列，或者记录错误日志。
• 数据类型转换： 示例中将键和值都转换为 String。如果 Kafka 记录的键或值是其他数据类型（例如 Avro、Protobuf、JSON），你需要在 deserialize 方法中引入相应的反序列化库和逻辑来解析 byte[]。
• 访问头部信息： ConsumerRecord 还提供了 headers() 方法来访问 Kafka 记录的头部信息。如果你的生产者在记录中添加了自定义头部，你也可以在 deserialize 方法中提取并处理它们。
• 性能考量： 自定义反序列化器会增加一些处理开销，但对于需要访问键或其他元数据的场景来说是必要的。对于高性能要求极高的场景，应确保反序列化逻辑尽可能高效。
• Flink 版本兼容性： 本教程基于 Flink 1.15+ 版本，使用了 KafkaSource API。如果你使用的是较旧的 Flink 版本（如 Flink 1.11 或更早），可能需要使用 FlinkKafkaConsumer，其配置方式略有不同，但核心思想（实现 DeserializationSchema）是相同的。

总结

通过实现自定义的 KafkaRecordDeserializationSchema，Apache Flink 能够灵活且全面地处理来自 Kafka 的复杂记录结构，包括带键的记录、时间戳以及其他重要的元数据。这种方法为构建功能强大、数据解析精细的 Flink 流处理应用程序提供了坚实的基础。理解并掌握自定义反序列化机制是开发高级 Flink-Kafka 集成应用的关键一步。