Flink 1.14.0 消费 kafka 数据自定义反序列化类

本文涉及的产品
实时计算 Flink 版,5000CU*H 3个月
简介: 在最近发布的 Flink 1.14.0 版本中对 Source 接口进行了重构,细节可以参考 FLIP-27: Refactor Source Interface重构之后 API 层面的改动还是非常大的,那在使用新的 API 消费 kafka 数据的时候如何自定义序列化类呢?

在最近发布的 Flink 1.14.0 版本中对 Source 接口进行了重构,细节可以参考 FLIP-27: Refactor Source Interface


重构之后 API 层面的改动还是非常大的,那在使用新的 API 消费 kafka 数据的时候如何自定义序列化类呢?


Kafka Source


KafkaSource<String> source = KafkaSource.<String>builder()
    .setBootstrapServers(brokers)
    .setTopics("input-topic")
    .setGroupId("my-group")
    .setStartingOffsets(OffsetsInitializer.earliest())
    .setValueOnlyDeserializer(new SimpleStringSchema())
    .build();
env.fromSource(source, WatermarkStrategy.noWatermarks(), "Kafka Source");


KafkaSourceBuilder 类提供了两个方法来反序列数据,分别是 setDeserializer 和 setValueOnlyDeserializer 从名字上就应该可以看出这两者的区别,前者是反序列化完整的 ConsumerRecord,后者只反序列化 ConsumerRecord 的 value.然后我们来看一下底层的源码


KafkaSourceBuilder 源码


/**
 * Sets the {@link KafkaRecordDeserializationSchema deserializer} of the {@link
 * org.apache.kafka.clients.consumer.ConsumerRecord ConsumerRecord} for KafkaSource.
 *
 * @param recordDeserializer the deserializer for Kafka {@link
 *     org.apache.kafka.clients.consumer.ConsumerRecord ConsumerRecord}.
 * @return this KafkaSourceBuilder.
 */
public KafkaSourceBuilder<OUT> setDeserializer(
        KafkaRecordDeserializationSchema<OUT> recordDeserializer) {
    this.deserializationSchema = recordDeserializer;
    return this;
}
/**
 * Sets the {@link KafkaRecordDeserializationSchema deserializer} of the {@link
 * org.apache.kafka.clients.consumer.ConsumerRecord ConsumerRecord} for KafkaSource. The given
 * {@link DeserializationSchema} will be used to deserialize the value of ConsumerRecord. The
 * other information (e.g. key) in a ConsumerRecord will be ignored.
 *
 * @param deserializationSchema the {@link DeserializationSchema} to use for deserialization.
 * @return this KafkaSourceBuilder.
 */
public KafkaSourceBuilder<OUT> setValueOnlyDeserializer(
        DeserializationSchema<OUT> deserializationSchema) {
    this.deserializationSchema =
            KafkaRecordDeserializationSchema.valueOnly(deserializationSchema);
    return this;
}


可以看到这两个方法实际上是一样的,虽然两个方法的参数不同,setDeserializer 方法参数类型是 KafkaRecordDeserializationSchema 而 setValueOnlyDeserializer 方法的参数类型是 DeserializationSchema 那这两种参数类型有什么区别和联系呢?下面会进一步解释, 但是这两个方法最后返回的都是 KafkaRecordDeserializationSchema 对象,我们继续来看 KafkaRecordDeserializationSchema 的源码


先来看一下 DeserializationSchema 的部分源码


DeserializationSchema 源码


@Public
public interface DeserializationSchema<T> extends Serializable, ResultTypeQueryable<T> {
    @PublicEvolving
    default void open(InitializationContext context) throws Exception {}
    T deserialize(byte[] message) throws IOException;
    @PublicEvolving
    default void deserialize(byte[] message, Collector<T> out) throws IOException {
        T deserialize = deserialize(message);
        if (deserialize != null) {
            out.collect(deserialize);
        }
    }
    boolean isEndOfStream(T nextElement);
}


KafkaRecordDeserializationSchema 源码


/** An interface for the deserialization of Kafka records. */
public interface KafkaRecordDeserializationSchema<T> extends Serializable, ResultTypeQueryable<T> {
    @PublicEvolving
    default void open(DeserializationSchema.InitializationContext context) throws Exception {}
    @PublicEvolving
    void deserialize(ConsumerRecord<byte[], byte[]> record, Collector<T> out) throws IOException;
    static <V> KafkaRecordDeserializationSchema<V> of(
            KafkaDeserializationSchema<V> kafkaDeserializationSchema) {
        return new KafkaDeserializationSchemaWrapper<>(kafkaDeserializationSchema);
    }
    static <V> KafkaRecordDeserializationSchema<V> valueOnly(
            DeserializationSchema<V> valueDeserializationSchema) {
        return new KafkaValueOnlyDeserializationSchemaWrapper<>(valueDeserializationSchema);
    }
    static <V> KafkaRecordDeserializationSchema<V> valueOnly(
            Class<? extends Deserializer<V>> valueDeserializerClass) {
        return new KafkaValueOnlyDeserializerWrapper<>(
                valueDeserializerClass, Collections.emptyMap());
    }
    static <V, D extends Configurable & Deserializer<V>>
            KafkaRecordDeserializationSchema<V> valueOnly(
                    Class<D> valueDeserializerClass, Map<String, String> config) {
        return new KafkaValueOnlyDeserializerWrapper<>(valueDeserializerClass, config);
    }
}


顾名思义,这两个都是反序列接口,并且都继承了 Serializable, ResultTypeQueryable 这两个接口。不同点是,deserialize 方法的参数不一样,KafkaDeserializationSchema 接口很明显是为反序列化 kafka 数据而生的。DeserializationSchema 接口可以反序列化任意二进制数据,更加具有通用性。所以这两个是同一级接口


如果你想要获取 kafka 的元数据信息选择实现 KafkaDeserializationSchema 接口就可以了,KafkaDeserializationSchema 接口还有 4 个静态方法,其中的 of 方法就是用来反序列化 ConsumerRecord 的,剩下的 3 个 valueOnly 是用来反序列化 kafka 消息中的 value 的.


到这里就非常清楚了,如果我们要自定义序列化类,实现 DeserializationSchema 和 KafkaRecordDeserializationSchema 任何一个都是可以的.下面就以 KafkaRecordDeserializationSchema 接口为例,实现一个简单的反序列化类.


MyKafkaDeserialization 自定义序列化类


package flink.stream.deserialization;
import bean.Jason;
import com.alibaba.fastjson.JSON;
import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.streaming.connectors.kafka.KafkaDeserializationSchema;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.log4j.Logger;
public class MyKafkaDeserialization implements KafkaDeserializationSchema<Jason> {
    private static final Logger log = Logger.getLogger(MyKafkaDeserialization.class);
    private final String encoding = "UTF8";
    private boolean includeTopic;
    private boolean includeTimestamp;
    public MyKafkaDeserialization(boolean includeTopic, boolean includeTimestamp) {
        this.includeTopic = includeTopic;
        this.includeTimestamp = includeTimestamp;
    }
    @Override
    public TypeInformation<Jason> getProducedType() {
        return TypeInformation.of(Jason.class);
    }
    @Override
    public boolean isEndOfStream(Jason nextElement) {
        return false;
    }
    @Override
    public Jason deserialize(ConsumerRecord<byte[], byte[]> consumerRecord) throws Exception {
        if (consumerRecord != null) {
            try {
                String value = new String(consumerRecord.value(), encoding);
                Jason jason = JSON.parseObject(value, Jason.class);
                if (includeTopic) jason.setTopic(consumerRecord.topic());
                if (includeTimestamp) jason.setTimestamp(consumerRecord.timestamp());
                return jason;
            } catch (Exception e) {
                log.error("deserialize failed : " + e.getMessage());
            }
        }
        return null;
    }
}


整个实现是非常简单的,这样就可以把消费到的数据反序列化成自己想要的格式,虽然 Flink 1.14.0 重构了 Source 接口,但是反序列化接口几乎没变,只不过在原有的基础上增加了几个方法而已.


使用


KafkaSource<Jason> source = KafkaSource.<Jason>builder()
        .setProperty("security.protocol", "SASL_PLAINTEXT")
        .setProperty("sasl.mechanism", "PLAIN")
        .setProperty("sasl.jaas.config", "org.apache.kafka.common.security.plain.PlainLoginModule required username=\"" + username + "\" password=\"" + password + "\";")
        // discover new partitions per 10 seconds
        .setProperty("partition.discovery.interval.ms", "10000")
        .setBootstrapServers(broker)
        .setTopics(topic)
        .setGroupId(group_id)
        .setStartingOffsets(OffsetsInitializer.earliest())
        .setDeserializer(KafkaRecordDeserializationSchema.of(new MyKafkaDeserialization(true, true)))
        // 只反序列化 value
        .setValueOnlyDeserializer(new MyDeSerializer())
        .build();


setDeserializer 和 setValueOnlyDeserializer 只用设置一个即可.

相关文章
|
2天前
|
SQL 关系型数据库 MySQL
实时计算 Flink版操作报错合集之从mysql读数据写到hive报错,是什么原因
在使用实时计算Flink版过程中,可能会遇到各种错误,了解这些错误的原因及解决方法对于高效排错至关重要。针对具体问题,查看Flink的日志是关键,它们通常会提供更详细的错误信息和堆栈跟踪,有助于定位问题。此外,Flink社区文档和官方论坛也是寻求帮助的好去处。以下是一些常见的操作报错及其可能的原因与解决策略。
|
2天前
|
SQL 流计算
实时计算 Flink版操作报错合集之怎么向一个未定义列的表中写入数据
在使用实时计算Flink版过程中,可能会遇到各种错误,了解这些错误的原因及解决方法对于高效排错至关重要。针对具体问题,查看Flink的日志是关键,它们通常会提供更详细的错误信息和堆栈跟踪,有助于定位问题。此外,Flink社区文档和官方论坛也是寻求帮助的好去处。以下是一些常见的操作报错及其可能的原因与解决策略。
|
2天前
|
Java 关系型数据库 数据库
实时计算 Flink版操作报错合集之拉取全量数据时,如何解决Checkpoint失败并且报错为 "java.lang.OutOfMemoryError: Java heap space"
在使用实时计算Flink版过程中,可能会遇到各种错误,了解这些错误的原因及解决方法对于高效排错至关重要。针对具体问题,查看Flink的日志是关键,它们通常会提供更详细的错误信息和堆栈跟踪,有助于定位问题。此外,Flink社区文档和官方论坛也是寻求帮助的好去处。以下是一些常见的操作报错及其可能的原因与解决策略。
|
2天前
|
分布式计算 Hadoop 关系型数据库
实时计算 Flink版操作报错合集之Hadoop在将文件写入HDFS时,无法在所有指定的数据节点上进行复制,该如何解决
在使用实时计算Flink版过程中,可能会遇到各种错误,了解这些错误的原因及解决方法对于高效排错至关重要。针对具体问题,查看Flink的日志是关键,它们通常会提供更详细的错误信息和堆栈跟踪,有助于定位问题。此外,Flink社区文档和官方论坛也是寻求帮助的好去处。以下是一些常见的操作报错及其可能的原因与解决策略。
|
2天前
|
SQL 分布式计算 Hadoop
实时计算 Flink版产品使用问题之建了一张upsert-kafka的flink表,但是数据为空,该如何排查
实时计算Flink版作为一种强大的流处理和批处理统一的计算框架,广泛应用于各种需要实时数据处理和分析的场景。实时计算Flink版通常结合SQL接口、DataStream API、以及与上下游数据源和存储系统的丰富连接器,提供了一套全面的解决方案,以应对各种实时计算需求。其低延迟、高吞吐、容错性强的特点,使其成为众多企业和组织实时数据处理首选的技术平台。以下是实时计算Flink版的一些典型使用合集。
|
2天前
|
Java 关系型数据库 数据库
实时计算 Flink版产品使用问题之如何将增量数据直接写入下游数据库
实时计算Flink版作为一种强大的流处理和批处理统一的计算框架,广泛应用于各种需要实时数据处理和分析的场景。实时计算Flink版通常结合SQL接口、DataStream API、以及与上下游数据源和存储系统的丰富连接器,提供了一套全面的解决方案,以应对各种实时计算需求。其低延迟、高吞吐、容错性强的特点,使其成为众多企业和组织实时数据处理首选的技术平台。以下是实时计算Flink版的一些典型使用合集。
|
3天前
|
消息中间件 SQL Kafka
实时计算 Flink版产品使用问题之如何实现两个阿里云账号下的Kafka进行数据的互相传输
实时计算Flink版作为一种强大的流处理和批处理统一的计算框架,广泛应用于各种需要实时数据处理和分析的场景。实时计算Flink版通常结合SQL接口、DataStream API、以及与上下游数据源和存储系统的丰富连接器,提供了一套全面的解决方案,以应对各种实时计算需求。其低延迟、高吞吐、容错性强的特点,使其成为众多企业和组织实时数据处理首选的技术平台。以下是实时计算Flink版的一些典型使用合集。
|
3天前
|
SQL 关系型数据库 MySQL
实时计算 Flink版产品使用问题之全量和增量同步数据的一致性、不丢失和不重复读取可以通过什么方式保证
实时计算Flink版作为一种强大的流处理和批处理统一的计算框架,广泛应用于各种需要实时数据处理和分析的场景。实时计算Flink版通常结合SQL接口、DataStream API、以及与上下游数据源和存储系统的丰富连接器,提供了一套全面的解决方案,以应对各种实时计算需求。其低延迟、高吞吐、容错性强的特点,使其成为众多企业和组织实时数据处理首选的技术平台。以下是实时计算Flink版的一些典型使用合集。
|
3天前
|
SQL 消息中间件 关系型数据库
实时计算 Flink版产品使用问题之元数据血缘可以通过什么来获取
实时计算Flink版作为一种强大的流处理和批处理统一的计算框架,广泛应用于各种需要实时数据处理和分析的场景。实时计算Flink版通常结合SQL接口、DataStream API、以及与上下游数据源和存储系统的丰富连接器,提供了一套全面的解决方案,以应对各种实时计算需求。其低延迟、高吞吐、容错性强的特点,使其成为众多企业和组织实时数据处理首选的技术平台。以下是实时计算Flink版的一些典型使用合集。
|
2天前
|
监控 关系型数据库 MySQL
实时计算 Flink版产品使用问题之在进行全量数据初始化时,连接器一般会采用什么策略
实时计算Flink版作为一种强大的流处理和批处理统一的计算框架,广泛应用于各种需要实时数据处理和分析的场景。实时计算Flink版通常结合SQL接口、DataStream API、以及与上下游数据源和存储系统的丰富连接器,提供了一套全面的解决方案,以应对各种实时计算需求。其低延迟、高吞吐、容错性强的特点,使其成为众多企业和组织实时数据处理首选的技术平台。以下是实时计算Flink版的一些典型使用合集。