基于Flink 1.11实现Kafka通用反序列化方案

在Flink 1.11版本中，通过Kafka消费记录时使用KafkaDeserializationSchema反序列化对象处理业务逻辑，但每次新增对象类型都需要创建对应的反序列化类，代码冗余严重。现有示例代码如下：

// 现有反序列化类
class ADeserialize extends KafkaDeserializationSchema[TypeAClass] {
  val mapper: ObjectMapper = new ObjectMapper

  override def isEndOfStream(nextElement: TypeAClass): Boolean = false

  override def deserialize(record: ConsumerRecord[Array[Byte], Array[Byte]]): TypeAClass = {
    mapper.readValue(record.value(), classOf[TypeAClass])
  }

  override def getProducedType: TypeInformation[TypeAClass] = Types.CASE_CLASS[TypeAClass]
}

// 新增的反序列化类
class BDeserialize extends KafkaDeserializationSchema[TypeBClass] {
  val mapper: ObjectMapper = new ObjectMapper

  override def isEndOfStream(nextElement: TypeBClass): Boolean = false

  override def deserialize(record: ConsumerRecord[Array[Byte], Array[Byte]]): TypeBClass = {
    mapper.readValue(record.value(), classOf[TypeBClass])
  }

  override def getProducedType: TypeInformation[TypeBClass] = Types.CASE_CLASS[TypeBClass]
}

尝试实现通用反序列化类但失败，寻求解决方案。

可以通过泛型类实现通用的Kafka反序列化器，核心是在构造时传入目标类型的Class对象，并利用Flink的TypeInformation工具生成对应的类型信息。同时注意配置Jackson的Scala模块，避免Scala case class序列化/反序列化时的兼容性问题。

通用反序列化类实现 #

import org.apache.flink.api.common.typeinfo.TypeInformation
import org.apache.flink.streaming.connectors.kafka.KafkaDeserializationSchema
import org.apache.kafka.clients.consumer.ConsumerRecord
import com.fasterxml.jackson.databind.ObjectMapper
import com.fasterxml.jackson.module.scala.DefaultScalaModule

class GenericKafkaDeserializer[T](targetClass: Class[T]) extends KafkaDeserializationSchema[T] {
  // 初始化ObjectMapper并注册Scala模块，支持case class
  private val mapper: ObjectMapper = new ObjectMapper()
    .registerModule(DefaultScalaModule)

  override def isEndOfStream(nextElement: T): Boolean = false

  override def deserialize(record: ConsumerRecord[Array[Byte], Array[Byte]]): T = {
    mapper.readValue(record.value(), targetClass)
  }

  override def getProducedType: TypeInformation[T] = {
    // 利用Flink的TypeInformation获取泛型类型信息
    TypeInformation.of(targetClass)
  }
}

使用方式 #

在创建Kafka数据源时，直接传入目标类型的Class对象即可：

import org.apache.flink.streaming.api.scala.StreamExecutionEnvironment
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer

val env = StreamExecutionEnvironment.getExecutionEnvironment

// 消费TypeAClass类型的数据
val consumerA = new FlinkKafkaConsumer[TypeAClass](
  "topic-a",
  new GenericKafkaDeserializer[TypeAClass](classOf[TypeAClass]),
  kafkaProperties
)

// 消费TypeBClass类型的数据
val consumerB = new FlinkKafkaConsumer[TypeBClass](
  "topic-b",
  new GenericKafkaDeserializer[TypeBClass](classOf[TypeBClass]),
  kafkaProperties
)

// 后续处理逻辑
env.addSource(consumerA).print()
env.execute("Generic Deserialization Demo")

注意事项 #

• Jackson Scala模块依赖：确保项目中包含Jackson Scala模块的依赖，Maven依赖示例：

<dependency>
  <groupId>com.fasterxml.jackson.module</groupId>
  <artifactId>jackson-module-scala_2.11</artifactId>
  <version>2.11.4</version> <!-- 版本需与Flink 1.11自带的Jackson版本匹配 -->
</dependency>

• TypeInformation兼容性：Flink 1.11中TypeInformation.of(Class)可正确识别Scala case class，普通场景下无需额外处理；若为复杂泛型类型，可能需要手动指定TypeInformation。
• ObjectMapper复用：将ObjectMapper声明为类的私有成员，避免每次反序列化都创建新实例，提升性能。

内容的提问来源于stack exchange，提问作者sclee1