图文精华

About云-梭伦科技»专题 › 技术学习(版主发帖区) › 大数据学习 › Flink › 技术应用› Flink 1.14.0 消费 kafka 数据自定义反序列化类
分享

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

Mirinda 2021-10-14 10:27:27 发表于 技术应用 [显示全部楼层] 回帖奖励 阅读模式 关闭右栏 0 2348
本帖最后由 Mirinda 于 2021-10-14 10:30 编辑

问题导读：
1.你有没有阅读源码的习惯？
2.有阅读过kafka源码吗？
3.你对文中的自定义序列化类有什么看法？

在最近发布的 Flink 1.14.0 版本中对 Source 接口进行了重构,细节可以参考 FLIP-27: Refactor Source Interface
重构之后 API 层面的改动还是非常大的,那在使用新的 API 消费 kafka 数据的时候如何自定义序列化类呢?
  1. Kafka SourceKafkaSource<String> source = KafkaSource.<String>builder()

  2.     .setBootstrapServers(brokers)

  3.     .setTopics("input-topic")

  4.     .setGroupId("my-group")

  5.     .setStartingOffsets(OffsetsInitializer.earliest())

  6.     .setValueOnlyDeserializer(new SimpleStringSchema())

  7.     .build();

  8. 

  9. env.fromSource(source, WatermarkStrategy.noWatermarks(), "Kafka Source");
复制代码

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

  2. * Sets the {@link KafkaRecordDeserializationSchema deserializer} of the {@link

  3. * org.apache.kafka.clients.consumer.ConsumerRecord ConsumerRecord} for KafkaSource.

  4. *

  5. * @param recordDeserializer the deserializer for Kafka {@link

  6. *     org.apache.kafka.clients.consumer.ConsumerRecord ConsumerRecord}.

  7. * @return this KafkaSourceBuilder.

  8. */

  9. public KafkaSourceBuilder<OUT> setDeserializer(

  10.         KafkaRecordDeserializationSchema<OUT> recordDeserializer) {

  11.     this.deserializationSchema = recordDeserializer;

  12.     return this;

  13. }

  14. 

  15. /**

  16. * Sets the {@link KafkaRecordDeserializationSchema deserializer} of the {@link

  17. * org.apache.kafka.clients.consumer.ConsumerRecord ConsumerRecord} for KafkaSource. The given

  18. * {@link DeserializationSchema} will be used to deserialize the value of ConsumerRecord. The

  19. * other information (e.g. key) in a ConsumerRecord will be ignored.

  20. *

  21. * @param deserializationSchema the {@link DeserializationSchema} to use for deserialization.

  22. * @return this KafkaSourceBuilder.

  23. */

  24. public KafkaSourceBuilder<OUT> setValueOnlyDeserializer(

  25.         DeserializationSchema<OUT> deserializationSchema) {

  26.     this.deserializationSchema =

  27.             KafkaRecordDeserializationSchema.valueOnly(deserializationSchema);

  28.     return this;

  29. }
复制代码

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

  2. public interface DeserializationSchema<T> extends Serializable, ResultTypeQueryable<T> {

  3.    

  4.     @PublicEvolving

  5.     default void open(InitializationContext context) throws Exception {}

  6.    

  7.     T deserialize(byte[] message) throws IOException;

  8. 

  9.     @PublicEvolving

  10.     default void deserialize(byte[] message, Collector<T> out) throws IOException {

  11.         T deserialize = deserialize(message);

  12.         if (deserialize != null) {

  13.             out.collect(deserialize);

  14.         }

  15.     }

  16.    

  17.     boolean isEndOfStream(T nextElement);

  18. }
复制代码

KafkaRecordDeserializationSchema 源码
  1. /** An interface for the deserialization of Kafka records. */

  2. public interface KafkaRecordDeserializationSchema<T> extends Serializable, ResultTypeQueryable<T> {

  3.    

  4.     @PublicEvolving

  5.     default void open(DeserializationSchema.InitializationContext context) throws Exception {}

  6. 

  7.     @PublicEvolving

  8.     void deserialize(ConsumerRecord<byte[], byte[]> record, Collector<T> out) throws IOException;

  9.   

  10.    

  11.     static <V> KafkaRecordDeserializationSchema<V> of(

  12.             KafkaDeserializationSchema<V> kafkaDeserializationSchema) {

  13.         return new KafkaDeserializationSchemaWrapper<>(kafkaDeserializationSchema);

  14.     }

  15. 

  16.    

  17.     static <V> KafkaRecordDeserializationSchema<V> valueOnly(

  18.             DeserializationSchema<V> valueDeserializationSchema) {

  19.         return new KafkaValueOnlyDeserializationSchemaWrapper<>(valueDeserializationSchema);

  20.     }

  21. 

  22.    

  23.     static <V> KafkaRecordDeserializationSchema<V> valueOnly(

  24.             Class<? extends Deserializer<V>> valueDeserializerClass) {

  25.         return new KafkaValueOnlyDeserializerWrapper<>(

  26.                 valueDeserializerClass, Collections.emptyMap());

  27.     }

  28. 

  29. 

  30.     static <V, D extends Configurable & Deserializer<V>>

  31.             KafkaRecordDeserializationSchema<V> valueOnly(

  32.                     Class<D> valueDeserializerClass, Map<String, String> config) {

  33.         return new KafkaValueOnlyDeserializerWrapper<>(valueDeserializerClass, config);

  34.     }

  35. }
复制代码

顾名思义,这两个都是反序列接口,并且都继承了 Serializable, ResultTypeQueryable 这两个接口。不同点是，deserialize 方法的参数不一样，KafkaDeserializationSchema 接口很明显是为反序列化 kafka 数据而生的。DeserializationSchema 接口可以反序列化任意二进制数据，更加具有通用性。所以这两个是同一级接口
如果你想要获取 kafka 的元数据信息选择实现 KafkaDeserializationSchema 接口就可以了,KafkaDeserializationSchema 接口还有 4 个静态方法,其中的 of 方法就是用来反序列化 ConsumerRecord 的,剩下的 3 个 valueOnly 是用来反序列化 kafka 消息中的 value 的.
到这里就非常清楚了,如果我们要自定义序列化类,实现 DeserializationSchema 和 KafkaRecordDeserializationSchema 任何一个都是可以的.下面就以 KafkaRecordDeserializationSchema 接口为例,实现一个简单的反序列化类.
MyKafkaDeserialization 自定义序列化类
  1. package flink.stream.deserialization;

  2. import bean.Jason;

  3. import com.alibaba.fastjson.JSON;

  4. import org.apache.flink.api.common.typeinfo.TypeInformation;

  5. import org.apache.flink.streaming.connectors.kafka.KafkaDeserializationSchema;

  6. import org.apache.kafka.clients.consumer.ConsumerRecord;

  7. import org.apache.log4j.Logger;

  8. 

  9. public class MyKafkaDeserialization implements KafkaDeserializationSchema<Jason> {

  10.     private static final Logger log = Logger.getLogger(MyKafkaDeserialization.class);

  11.     private final String encoding = "UTF8";

  12.     private boolean includeTopic;

  13.     private boolean includeTimestamp;

  14. 

  15.     public MyKafkaDeserialization(boolean includeTopic, boolean includeTimestamp) {

  16.         this.includeTopic = includeTopic;

  17.         this.includeTimestamp = includeTimestamp;

  18.     }

  19. 

  20.     @Override

  21.     public TypeInformation<Jason> getProducedType() {

  22.         return TypeInformation.of(Jason.class);

  23.     }

  24. 

  25.     @Override

  26.     public boolean isEndOfStream(Jason nextElement) {

  27.         return false;

  28.     }

  29. 

  30.     @Override

  31.     public Jason deserialize(ConsumerRecord<byte[], byte[]> consumerRecord) throws Exception {

  32.         if (consumerRecord != null) {

  33.             try {

  34.                 String value = new String(consumerRecord.value(), encoding);

  35.                 Jason jason = JSON.parseObject(value, Jason.class);

  36.                 if (includeTopic) jason.setTopic(consumerRecord.topic());

  37.                 if (includeTimestamp) jason.setTimestamp(consumerRecord.timestamp());

  38.                 return jason;

  39.             } catch (Exception e) {

  40.                 log.error("deserialize failed : " + e.getMessage());

  41.             }

  42.         }

  43.         return null;

  44.     }

  45. }
复制代码

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

  2.         .setProperty("security.protocol", "SASL_PLAINTEXT")

  3.         .setProperty("sasl.mechanism", "PLAIN")

  4.         .setProperty("sasl.jaas.config", "org.apache.kafka.common.security.plain.PlainLoginModule required username="" + username + "" password="" + password + "";")

  5.         // discover new partitions per 10 seconds

  6.         .setProperty("partition.discovery.interval.ms", "10000")

  7.         .setBootstrapServers(broker)

  8.         .setTopics(topic)

  9.         .setGroupId(group_id)

  10.         .setStartingOffsets(OffsetsInitializer.earliest())

  11.         .setDeserializer(KafkaRecordDeserializationSchema.of(new MyKafkaDeserialization(true, true)))

  12.         // 只反序列化 value

  13.         .setValueOnlyDeserializer(new MyDeSerializer())

  14.         .build();
复制代码

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

推荐阅读
Flink 任务实时监控最佳实践
Flink on yarn 实时日志收集最佳实践
Flink 1.14.0 全新的 Kafka Connector



最新经典文章，欢迎关注公众号


回复

使用道具 举报

自定义广告语

没找到任何评论，期待你打破沉寂

高级模式
B Color Image Link Quote Code Smilies
您需要登录后才可以回帖 登录 | 立即注册

本版积分规则

发表新帖
Mirinda

超级版主

关注

  • 123

    主题

  • 125

    帖子

  • 11

    粉丝

TA的主题
关闭

推荐上一条 /2 下一条

关于我们· 联系我们· 加入我们· 小黑屋· 合作伙伴   

Copyright © 2001-2025 About云-梭伦科技 Powered by Discuz! X3.4 Licensed Discuz Team.

简书 / Smrz 京ICP备2020039040号 Wxb 简书网举报电话：021-34700000